Function TestBackendScheduler

modules/backendscheduler/backendscheduler_test.go:33–241

(t *testing.T)

var tenant = "test-tenant"

func TestBackendScheduler(t *testing.T) {
	cfg := Config{}
	cfg.RegisterFlagsAndApplyDefaults("", &flag.FlagSet{})
	cfg.BackendFlushInterval = 100 * time.Millisecond

	tmpDir := t.TempDir()
	cfg.LocalWorkPath = tmpDir

	var (
		ctx, cancel   = context.WithCancel(context.Background())
		store, rr, ww = newStore(ctx, t, tmpDir)
	)

	defer func() {
		cancel()
		store.Shutdown()
	}()

	limits, err := overrides.NewOverrides(overrides.Config{Defaults: overrides.Overrides{}}, nil, prometheus.DefaultRegisterer)
	require.NoError(t, err)

	t.Run("next with no jobs returns correct errors", func(t *testing.T) {
		s, err := New(cfg, store, limits, rr, ww)
		require.NoError(t, err)

		resp, err := s.Next(ctx, &tempopb.NextJobRequest{
			WorkerId: "test-worker",
		})
		require.Error(t, err)
		errStatus, ok := status.FromError(err)
		require.True(t, ok)
		require.Equal(t, errStatus.Code(), codes.NotFound)

		require.NotNil(t, resp)
		require.Equal(t, "", resp.JobId)
	})

	tenantCount := 5
	tenantBlockIDs := make(map[string][]backend.UUID)

	// Push some data to a few tenants
	for i := range tenantCount {
		testTenant := tenant + strconv.Itoa(i)
		tenantBlockIDs[testTenant] = writeTenantBlocks(ctx, t, backend.NewWriter(ww), testTenant, 10)
	}

	time.Sleep(500 * time.Millisecond)

	t.Run("jobs need doing", func(t *testing.T) {
		s, err := New(cfg, store, limits, rr, ww)
		require.NoError(t, err)

		// Start the scheduler
		err = s.starting(ctx)
		require.NoError(t, err)

		go func() {
			// Start a goroutine to run the service