Function Test_instance_concurrency

modules/generator/instance_test.go:32–88

(t *testing.T)

)

func Test_instance_concurrency(t *testing.T) {
	// Both instances use the same overrides, this map will be accessed by both
	overrides := &mockOverrides{}
	overrides.processors = map[string]struct{}{
		processor.SpanMetricsName:   {},
		processor.ServiceGraphsName: {},
	}
	cfg := &Config{}
	cfg.RegisterFlagsAndApplyDefaults("", &flag.FlagSet{})

	instance1, err := newInstance(cfg, "test", overrides, &noopStorage{}, log.NewNopLogger())
	assert.NoError(t, err)

	instance2, err := newInstance(cfg, "test", overrides, &noopStorage{}, log.NewNopLogger())
	assert.NoError(t, err)

	end := make(chan struct{})

	accessor := func(f func()) {
		for {
			select {
			case <-end:
				return
			default:
				f()
			}
		}
	}

	go accessor(func() {
		req := test.MakeBatch(1, nil)
		instance1.pushSpans(context.Background(), &tempopb.PushSpansRequest{Batches: []*v1.ResourceSpans{req}})
	})

	go accessor(func() {
		req := test.MakeBatch(1, nil)
		instance2.pushSpans(context.Background(), &tempopb.PushSpansRequest{Batches: []*v1.ResourceSpans{req}})
	})

	go accessor(func() {
		err := instance1.updateProcessors()
		assert.NoError(t, err)
	})

	go accessor(func() {
		err := instance2.updateProcessors()
		assert.NoError(t, err)
	})

	time.Sleep(100 * time.Millisecond)

	instance1.shutdown()
	instance2.shutdown()

	time.Sleep(10 * time.Millisecond)
	close(end)
}