mirror of
https://github.com/hanzoai/traces.git
synced 2026-10-30 18:20:41 +00:00
Zach LeslieandGitHub
c5fe2e7e6b
* [backend-scheduler] initial work file sharding for more targeted marshaling * Add integration coverage for both work implementations * Update config manifest * Add integration coverage for sharded work configuration * Fix integration test config path * Add test coverage for sharded work migrations and marshaling * Use new sharded work file in the backend and handle migration * Clean up backend after migration * Use shared method for shard file name * Add method to write files atomically and test * Update work interface for simpler flush/load interactions * Update callers for local flush/load interface methods * Fix lint * Fix lint * Fix lint * Clean up * Clean up after refactor new interface functions * Promote the sharded implementation to be the only implementation * Clean up comments * Generate config manifest * Clean up interface and improve test coverage * Drop transitional comments * Protect work marshal/unmarshal with a mutex * Add test coverage for concurrent marshal/unmarshal * Lint * Fix race condition in Marshal/Unmarshal when jobs are operating * Relocate methods * Update changelog * Drop interface method only used in tests * Rename method for clarity, add doc * Update doc for clarity * Lock for safety * Clarify and improve locking flush behavior * Encapsulate the lock/unlock in a func * Address some PR feedback * Further unifi shard file name generation * Modify work to use int rather than uint8 * Fix lint
93 lines
2.4 KiB
Go
93 lines
2.4 KiB
Go
package backendscheduler
|
|
|
|
import (
|
|
"context"
|
|
"flag"
|
|
"os"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/google/uuid"
|
|
"github.com/grafana/tempo/modules/backendscheduler/work"
|
|
"github.com/grafana/tempo/modules/overrides"
|
|
"github.com/grafana/tempo/tempodb/backend"
|
|
"github.com/prometheus/client_golang/prometheus"
|
|
"github.com/stretchr/testify/require"
|
|
)
|
|
|
|
func TestCache(t *testing.T) {
|
|
tmpDir := t.TempDir()
|
|
cfg := Config{}
|
|
cfg.RegisterFlagsAndApplyDefaults("", &flag.FlagSet{})
|
|
cfg.ProviderConfig.Retention.Interval = 100 * time.Millisecond
|
|
cfg.LocalWorkPath = tmpDir + "/work"
|
|
|
|
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)
|
|
|
|
s, err := New(cfg, store, limits, rr, ww)
|
|
require.NoError(t, err)
|
|
|
|
// No file should exist yet
|
|
err = s.work.LoadFromLocal(ctx, s.cfg.LocalWorkPath)
|
|
require.NoError(t, err)
|
|
|
|
// Flush the empty work cache
|
|
err = s.work.FlushToLocal(ctx, s.cfg.LocalWorkPath, nil)
|
|
require.NoError(t, err)
|
|
|
|
// No error loading the empty work cache
|
|
err = s.work.LoadFromLocal(ctx, s.cfg.LocalWorkPath)
|
|
require.NoError(t, err)
|
|
|
|
u := uuid.NewString()
|
|
// Add a job to the work TestCache
|
|
err = s.work.AddJob(&work.Job{
|
|
ID: u,
|
|
})
|
|
require.NoError(t, err)
|
|
|
|
// Flush the work cache to the backend
|
|
err = s.flushWorkCacheToBackend(ctx)
|
|
require.NoError(t, err)
|
|
|
|
os.Remove(s.cfg.LocalWorkPath + "/" + backend.WorkFileName)
|
|
|
|
// Test backend fallback
|
|
err = s.loadWorkCacheFromBackend(ctx)
|
|
require.NoError(t, err)
|
|
|
|
job := s.work.GetJob(u)
|
|
require.NotNil(t, job, "Job should be found in the work cache")
|
|
|
|
jobs := s.work.ListJobs()
|
|
require.Len(t, jobs, 1, "There should be one job in the work cache")
|
|
require.Equal(t, u, jobs[0].ID, "Job ID should match the one added")
|
|
|
|
// Create a new scheduler with a different local work path
|
|
cfg.LocalWorkPath = tmpDir + "/work2"
|
|
s, err = New(cfg, store, limits, rr, ww)
|
|
require.NoError(t, err)
|
|
|
|
// Test loading from backend when no local cache exists
|
|
err = s.loadWorkCacheFromBackend(ctx)
|
|
require.NoError(t, err)
|
|
|
|
job = s.work.GetJob(u)
|
|
require.NotNil(t, job, "Job should be found in the work cache")
|
|
|
|
jobs = s.work.ListJobs()
|
|
require.Len(t, jobs, 1, "There should be one job in the work cache")
|
|
require.Equal(t, u, jobs[0].ID, "Job ID should match the one added")
|
|
}
|