diff options
| -rw-r--r-- | go.mod | 2 | ||||
| -rw-r--r-- | go.sum | 4 | ||||
| -rw-r--r-- | internal/cli/gitaly/serve.go | 4 | ||||
| -rw-r--r-- | internal/git/housekeeping/clean_stale_data.go | 15 | ||||
| -rw-r--r-- | internal/git/housekeeping/clean_stale_data_test.go | 22 | ||||
| -rw-r--r-- | internal/git/housekeeping/manager.go | 7 | ||||
| -rw-r--r-- | internal/git/housekeeping/optimize_repository.go | 10 | ||||
| -rw-r--r-- | internal/git/housekeeping/optimize_repository_ext_test.go | 5 | ||||
| -rw-r--r-- | internal/git/housekeeping/optimize_repository_test.go | 39 | ||||
| -rw-r--r-- | internal/git/objectpool/fetch.go | 4 | ||||
| -rw-r--r-- | internal/gitaly/maintenance/optimize.go | 13 | ||||
| -rw-r--r-- | internal/gitaly/maintenance/optimize_test.go | 9 | ||||
| -rw-r--r-- | internal/gitaly/service/ref/find_local_branches_test.go | 39 | ||||
| -rw-r--r-- | internal/gitaly/service/ref/util.go | 35 | ||||
| -rw-r--r-- | internal/gitaly/service/repository/optimize.go | 3 | ||||
| -rw-r--r-- | internal/gitaly/service/repository/optimize_test.go | 3 | ||||
| -rw-r--r-- | internal/gitaly/storage/storagemgr/transaction_manager.go | 2 |
17 files changed, 141 insertions, 75 deletions
@@ -45,7 +45,7 @@ require ( golang.org/x/text v0.13.0 golang.org/x/time v0.3.0 google.golang.org/genproto/googleapis/rpc v0.0.0-20230822172742-b8732ec3820d - google.golang.org/grpc v1.57.0 + google.golang.org/grpc v1.58.0 google.golang.org/protobuf v1.31.0 ) @@ -1052,8 +1052,8 @@ google.golang.org/grpc v1.38.0/go.mod h1:NREThFqKR1f3iQ6oBuvc5LadQuXVGo9rkm5ZGrQ google.golang.org/grpc v1.39.0/go.mod h1:PImNr+rS9TWYb2O4/emRugxiyHZ5JyHW5F+RPnDzfrE= google.golang.org/grpc v1.39.1/go.mod h1:PImNr+rS9TWYb2O4/emRugxiyHZ5JyHW5F+RPnDzfrE= google.golang.org/grpc v1.45.0/go.mod h1:lN7owxKUQEqMfSyQikvvk5tf/6zMPsrK+ONuO11+0rQ= -google.golang.org/grpc v1.57.0 h1:kfzNeI/klCGD2YPMUlaGNT3pxvYfga7smW3Vth8Zsiw= -google.golang.org/grpc v1.57.0/go.mod h1:Sd+9RMTACXwmub0zcNY2c4arhtrbBYD1AUHI/dt16Mo= +google.golang.org/grpc v1.58.0 h1:32JY8YpPMSR45K+c3o6b8VL73V+rR8k+DeMIr4vRH8o= +google.golang.org/grpc v1.58.0/go.mod h1:tgX3ZQDlNJGU96V6yHh1T/JeoBQ2TXdr43YbYSsCJk0= google.golang.org/grpc/cmd/protoc-gen-go-grpc v1.1.0/go.mod h1:6Kw0yEErY5E/yWrBtf03jp27GLLJujG4z/JK95pnjjw= google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8= google.golang.org/protobuf v0.0.0-20200221191635-4d8936d0db64/go.mod h1:kwYJMbMJ01Woi6D6+Kah6886xMZcty6N08ah7+eCXa0= diff --git a/internal/cli/gitaly/serve.go b/internal/cli/gitaly/serve.go index c45cfc262..1fa46a3d5 100644 --- a/internal/cli/gitaly/serve.go +++ b/internal/cli/gitaly/serve.go @@ -422,8 +422,8 @@ func run(cfg config.Cfg, logger logrus.FieldLogger) error { shutdownWorkers, err := maintenance.StartWorkers( ctx, logger, - maintenance.DailyOptimizationWorker(cfg, maintenance.OptimizerFunc(func(ctx context.Context, repo storage.Repository) error { - return housekeepingManager.OptimizeRepository(ctx, localrepo.New(locator, gitCmdFactory, catfileCache, repo)) + maintenance.DailyOptimizationWorker(cfg, maintenance.OptimizerFunc(func(ctx context.Context, logger logrus.FieldLogger, repo storage.Repository) error { + return housekeepingManager.OptimizeRepository(ctx, logger, localrepo.New(locator, gitCmdFactory, catfileCache, repo)) })), ) if err != nil { diff --git a/internal/git/housekeeping/clean_stale_data.go b/internal/git/housekeeping/clean_stale_data.go index 89eb30506..d8436b3d7 100644 --- a/internal/git/housekeeping/clean_stale_data.go +++ b/internal/git/housekeeping/clean_stale_data.go @@ -11,8 +11,7 @@ import ( "strings" "time" - "github.com/grpc-ecosystem/go-grpc-middleware/logging/logrus/ctxlogrus" - log "github.com/sirupsen/logrus" + "github.com/sirupsen/logrus" "gitlab.com/gitlab-org/gitaly/v16/internal/git" "gitlab.com/gitlab-org/gitaly/v16/internal/git/localrepo" "gitlab.com/gitlab-org/gitaly/v16/internal/gitaly/transaction" @@ -91,13 +90,13 @@ func DefaultStaleDataCleanup() CleanStaleDataConfig { } // CleanStaleData removes any stale data in the repository as per the provided configuration. -func (m *RepositoryManager) CleanStaleData(ctx context.Context, repo *localrepo.Repo, cfg CleanStaleDataConfig) error { +func (m *RepositoryManager) CleanStaleData(ctx context.Context, logger logrus.FieldLogger, repo *localrepo.Repo, cfg CleanStaleDataConfig) error { span, ctx := tracing.StartSpanIfHasParent(ctx, "housekeeping.CleanStaleData", nil) defer span.Finish() repoPath, err := repo.Path() if err != nil { - myLogger(ctx).WithError(err).Warn("housekeeping failed to get repo path") + myLogger(logger).WithError(err).Warn("housekeeping failed to get repo path") if structerr.GRPCCode(err) == codes.NotFound { return nil } @@ -110,7 +109,7 @@ func (m *RepositoryManager) CleanStaleData(ctx context.Context, repo *localrepo. return } - logEntry := myLogger(ctx) + logEntry := myLogger(logger) for staleDataType, count := range staleDataByType { logEntry = logEntry.WithField(fmt.Sprintf("stale_data.%s", staleDataType), count) m.prunedFilesTotal.WithLabelValues(staleDataType).Add(float64(count)) @@ -135,7 +134,7 @@ func (m *RepositoryManager) CleanStaleData(ctx context.Context, repo *localrepo. continue } staleDataByType["failures"]++ - myLogger(ctx).WithError(err).WithField("path", path).Warn("unable to remove stale file") + myLogger(logger).WithError(err).WithField("path", path).Warn("unable to remove stale file") } } @@ -657,6 +656,6 @@ func removeEmptyDirs(ctx context.Context, target string) (int, error) { return prunedDirsTotal + 1, nil } -func myLogger(ctx context.Context) *log.Entry { - return ctxlogrus.Extract(ctx).WithField("system", "housekeeping") +func myLogger(logger logrus.FieldLogger) logrus.FieldLogger { + return logger.WithField("system", "housekeeping") } diff --git a/internal/git/housekeeping/clean_stale_data_test.go b/internal/git/housekeeping/clean_stale_data_test.go index e86cc55eb..8ca4a0e83 100644 --- a/internal/git/housekeeping/clean_stale_data_test.go +++ b/internal/git/housekeeping/clean_stale_data_test.go @@ -401,7 +401,7 @@ func TestRepositoryManager_CleanStaleData(t *testing.T) { mgr := NewManager(cfg.Prometheus, nil) - require.NoError(t, mgr.CleanStaleData(ctx, repo, DefaultStaleDataCleanup())) + require.NoError(t, mgr.CleanStaleData(ctx, testhelper.SharedLogger(t), repo, DefaultStaleDataCleanup())) for _, e := range tc.entries { e.validate(t, repoPath) @@ -506,7 +506,7 @@ func TestRepositoryManager_CleanStaleData_references(t *testing.T) { mgr := NewManager(cfg.Prometheus, nil) - require.NoError(t, mgr.CleanStaleData(ctx, repo, DefaultStaleDataCleanup())) + require.NoError(t, mgr.CleanStaleData(ctx, testhelper.SharedLogger(t), repo, DefaultStaleDataCleanup())) var actual []string require.NoError(t, filepath.Walk(filepath.Join(repoPath, "refs"), func(path string, info os.FileInfo, _ error) error { @@ -633,7 +633,7 @@ func TestRepositoryManager_CleanStaleData_emptyRefDirs(t *testing.T) { mgr := NewManager(cfg.Prometheus, nil) - require.NoError(t, mgr.CleanStaleData(ctx, repo, DefaultStaleDataCleanup())) + require.NoError(t, mgr.CleanStaleData(ctx, testhelper.SharedLogger(t), repo, DefaultStaleDataCleanup())) for _, e := range tc.entries { e.validate(t, repoPath) @@ -769,7 +769,7 @@ func TestRepositoryManager_CleanStaleData_withSpecificFile(t *testing.T) { repo := localrepo.NewTestRepo(t, cfg, repoProto) mgr := NewManager(cfg.Prometheus, nil) - require.NoError(t, mgr.CleanStaleData(ctx, repo, DefaultStaleDataCleanup())) + require.NoError(t, mgr.CleanStaleData(ctx, testhelper.SharedLogger(t), repo, DefaultStaleDataCleanup())) for _, subcase := range []struct { desc string entry entry @@ -813,7 +813,7 @@ func TestRepositoryManager_CleanStaleData_withSpecificFile(t *testing.T) { require.NoError(t, err) require.ElementsMatch(t, subcase.expectedFiles, staleFiles) - require.NoError(t, mgr.CleanStaleData(ctx, repo, DefaultStaleDataCleanup())) + require.NoError(t, mgr.CleanStaleData(ctx, testhelper.SharedLogger(t), repo, DefaultStaleDataCleanup())) entry.validate(t, repoPath) }) @@ -867,7 +867,7 @@ func TestRepositoryManager_CleanStaleData_serverInfo(t *testing.T) { mgr := NewManager(cfg.Prometheus, nil) - require.NoError(t, mgr.CleanStaleData(ctx, repo, DefaultStaleDataCleanup())) + require.NoError(t, mgr.CleanStaleData(ctx, testhelper.SharedLogger(t), repo, DefaultStaleDataCleanup())) for _, entry := range entries { entry.validate(t, repoPath) @@ -997,7 +997,7 @@ func TestRepositoryManager_CleanStaleData_referenceLocks(t *testing.T) { mgr := NewManager(cfg.Prometheus, nil) - require.NoError(t, mgr.CleanStaleData(ctx, repo, tc.cfg)) + require.NoError(t, mgr.CleanStaleData(ctx, testhelper.SharedLogger(t), repo, tc.cfg)) for _, e := range tc.entries { e.validate(t, repoPath) @@ -1110,7 +1110,7 @@ func TestRepositoryManager_CleanStaleData_missingRepo(t *testing.T) { require.NoError(t, os.RemoveAll(repoPath)) - require.NoError(t, NewManager(cfg.Prometheus, nil).CleanStaleData(ctx, repo, DefaultStaleDataCleanup())) + require.NoError(t, NewManager(cfg.Prometheus, nil).CleanStaleData(ctx, testhelper.SharedLogger(t), repo, DefaultStaleDataCleanup())) } func TestRepositoryManager_CleanStaleData_unsetConfiguration(t *testing.T) { @@ -1151,7 +1151,7 @@ func TestRepositoryManager_CleanStaleData_unsetConfiguration(t *testing.T) { mgr := NewManager(cfg.Prometheus, nil) - require.NoError(t, mgr.CleanStaleData(ctx, repo, DefaultStaleDataCleanup())) + require.NoError(t, mgr.CleanStaleData(ctx, testhelper.SharedLogger(t), repo, DefaultStaleDataCleanup())) require.Equal(t, `[core] repositoryformatversion = 0 @@ -1190,7 +1190,7 @@ func TestRepositoryManager_CleanStaleData_unsetConfigurationTransactional(t *tes AuthInfo: backchannel.WithID(nil, 1234), }) - require.NoError(t, NewManager(cfg.Prometheus, txManager).CleanStaleData(ctx, repo, DefaultStaleDataCleanup())) + require.NoError(t, NewManager(cfg.Prometheus, txManager).CleanStaleData(ctx, testhelper.SharedLogger(t), repo, DefaultStaleDataCleanup())) require.Equal(t, 2, len(txManager.Votes())) configKeys := gittest.Exec(t, cfg, "-C", repoPath, "config", "--list", "--local", "--name-only") @@ -1242,7 +1242,7 @@ func TestRepositoryManager_CleanStaleData_pruneEmptyConfigSections(t *testing.T) mgr := NewManager(cfg.Prometheus, nil) - require.NoError(t, mgr.CleanStaleData(ctx, repo, DefaultStaleDataCleanup())) + require.NoError(t, mgr.CleanStaleData(ctx, testhelper.SharedLogger(t), repo, DefaultStaleDataCleanup())) require.Equal(t, `[core] repositoryformatversion = 0 filemode = true diff --git a/internal/git/housekeeping/manager.go b/internal/git/housekeeping/manager.go index d73152182..c92d0c297 100644 --- a/internal/git/housekeeping/manager.go +++ b/internal/git/housekeeping/manager.go @@ -6,6 +6,7 @@ import ( "time" "github.com/prometheus/client_golang/prometheus" + "github.com/sirupsen/logrus" "gitlab.com/gitlab-org/gitaly/v16/internal/git/localrepo" gitalycfgprom "gitlab.com/gitlab-org/gitaly/v16/internal/gitaly/config/prometheus" "gitlab.com/gitlab-org/gitaly/v16/internal/gitaly/transaction" @@ -15,10 +16,10 @@ import ( // such as the cleanup of unneeded files and optimizations for the repository's data structures. type Manager interface { // CleanStaleData removes any stale data in the repository as per the provided configuration. - CleanStaleData(ctx context.Context, repo *localrepo.Repo, cfg CleanStaleDataConfig) error + CleanStaleData(context.Context, logrus.FieldLogger, *localrepo.Repo, CleanStaleDataConfig) error // OptimizeRepository optimizes the repository's data structures such that it can be more // efficiently served. - OptimizeRepository(context.Context, *localrepo.Repo, ...OptimizeRepositoryOption) error + OptimizeRepository(context.Context, logrus.FieldLogger, *localrepo.Repo, ...OptimizeRepositoryOption) error // AddPackRefsInhibitor allows clients to block housekeeping from running git-pack-refs(1). AddPackRefsInhibitor(ctx context.Context, repoPath string) (bool, func(), error) } @@ -213,7 +214,7 @@ type RepositoryManager struct { dataStructureCount *prometheus.HistogramVec dataStructureSize *prometheus.HistogramVec dataStructureTimeSinceLastOptimization *prometheus.HistogramVec - optimizeFunc func(context.Context, *RepositoryManager, *localrepo.Repo, OptimizationStrategy) error + optimizeFunc func(context.Context, *RepositoryManager, logrus.FieldLogger, *localrepo.Repo, OptimizationStrategy) error repositoryStates repositoryStates } diff --git a/internal/git/housekeeping/optimize_repository.go b/internal/git/housekeeping/optimize_repository.go index ec010d33a..8fe65f594 100644 --- a/internal/git/housekeeping/optimize_repository.go +++ b/internal/git/housekeeping/optimize_repository.go @@ -7,8 +7,8 @@ import ( "strconv" "time" - "github.com/grpc-ecosystem/go-grpc-middleware/logging/logrus/ctxlogrus" "github.com/prometheus/client_golang/prometheus" + "github.com/sirupsen/logrus" "gitlab.com/gitlab-org/gitaly/v16/internal/git" "gitlab.com/gitlab-org/gitaly/v16/internal/git/localrepo" "gitlab.com/gitlab-org/gitaly/v16/internal/git/stats" @@ -41,6 +41,7 @@ func WithOptimizationStrategyConstructor(strategyConstructor OptimizationStrateg // or not depends on a set of heuristics. func (m *RepositoryManager) OptimizeRepository( ctx context.Context, + logger logrus.FieldLogger, repo *localrepo.Repo, opts ...OptimizeRepositoryOption, ) error { @@ -83,7 +84,7 @@ func (m *RepositoryManager) OptimizeRepository( strategy = cfg.StrategyConstructor(repositoryInfo) } - return m.optimizeFunc(ctx, m, repo, strategy) + return m.optimizeFunc(ctx, m, logger, repo, strategy) } func (m *RepositoryManager) reportRepositoryInfo(ctx context.Context, info stats.RepositoryInfo) { @@ -139,6 +140,7 @@ func (m *RepositoryManager) reportDataStructureSize(dataStructure string, size u func optimizeRepository( ctx context.Context, m *RepositoryManager, + logger logrus.FieldLogger, repo *localrepo.Repo, strategy OptimizationStrategy, ) error { @@ -148,7 +150,7 @@ func optimizeRepository( optimizations := make(map[string]string) defer func() { totalTimer.ObserveDuration() - ctxlogrus.Extract(ctx).WithField("optimizations", optimizations).Info("optimized repository") + logger.WithField("optimizations", optimizations).Info("optimized repository") for task, status := range optimizations { m.tasksTotal.WithLabelValues(task, status).Inc() @@ -158,7 +160,7 @@ func optimizeRepository( }() timer := prometheus.NewTimer(m.tasksLatency.WithLabelValues("clean-stale-data")) - if err := m.CleanStaleData(ctx, repo, DefaultStaleDataCleanup()); err != nil { + if err := m.CleanStaleData(ctx, logger, repo, DefaultStaleDataCleanup()); err != nil { return fmt.Errorf("could not execute houskeeping: %w", err) } timer.ObserveDuration() diff --git a/internal/git/housekeeping/optimize_repository_ext_test.go b/internal/git/housekeeping/optimize_repository_ext_test.go index 5b99a3325..b5060f755 100644 --- a/internal/git/housekeeping/optimize_repository_ext_test.go +++ b/internal/git/housekeeping/optimize_repository_ext_test.go @@ -7,8 +7,6 @@ import ( "testing" "time" - "github.com/grpc-ecosystem/go-grpc-middleware/logging/logrus/ctxlogrus" - "github.com/sirupsen/logrus" "github.com/sirupsen/logrus/hooks/test" "github.com/stretchr/testify/require" "gitlab.com/gitlab-org/gitaly/v16/internal/git/gittest" @@ -173,9 +171,8 @@ func TestPruneIfNeeded(t *testing.T) { } logger, hook := test.NewNullLogger() - ctx := ctxlogrus.ToContext(ctx, logrus.NewEntry(logger)) - require.NoError(t, housekeeping.NewManager(cfg.Prometheus, nil).OptimizeRepository(ctx, repo)) + require.NoError(t, housekeeping.NewManager(cfg.Prometheus, nil).OptimizeRepository(ctx, logger, repo)) require.Equal(t, tc.expectedLogEntries, hook.Entries[len(hook.Entries)-1].Data["optimizations"]) }) } diff --git a/internal/git/housekeeping/optimize_repository_test.go b/internal/git/housekeeping/optimize_repository_test.go index 0dd2d05fd..a1173334f 100644 --- a/internal/git/housekeeping/optimize_repository_test.go +++ b/internal/git/housekeeping/optimize_repository_test.go @@ -12,6 +12,7 @@ import ( "time" "github.com/prometheus/client_golang/prometheus/testutil" + "github.com/sirupsen/logrus" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "gitlab.com/gitlab-org/gitaly/v16/internal/command" @@ -1027,7 +1028,7 @@ func TestOptimizeRepository(t *testing.T) { manager := NewManager(cfg.Prometheus, txManager) - err := manager.OptimizeRepository(ctx, setup.repo, setup.options...) + err := manager.OptimizeRepository(ctx, testhelper.SharedLogger(t), setup.repo, setup.options...) require.Equal(t, setup.expectedErr, err) expectedMetrics := setup.expectedMetrics @@ -1075,7 +1076,7 @@ func TestOptimizeRepository_ConcurrencyLimit(t *testing.T) { repo := localrepo.NewTestRepo(t, cfg, repoProto) manager := NewManager(gitalycfgprom.Config{}, nil) - manager.optimizeFunc = func(context.Context, *RepositoryManager, *localrepo.Repo, OptimizationStrategy) error { + manager.optimizeFunc = func(context.Context, *RepositoryManager, logrus.FieldLogger, *localrepo.Repo, OptimizationStrategy) error { reqReceivedCh <- struct{}{} ch <- struct{}{} @@ -1083,14 +1084,14 @@ func TestOptimizeRepository_ConcurrencyLimit(t *testing.T) { } go func() { - require.NoError(t, manager.OptimizeRepository(ctx, repo)) + require.NoError(t, manager.OptimizeRepository(ctx, testhelper.SharedLogger(t), repo)) }() <-reqReceivedCh // When repository optimizations are performed for a specific repository already, // then any subsequent calls to the same repository should just return immediately // without doing any optimizations at all. - require.NoError(t, manager.OptimizeRepository(ctx, repo)) + require.NoError(t, manager.OptimizeRepository(ctx, testhelper.SharedLogger(t), repo)) <-ch }) @@ -1107,7 +1108,7 @@ func TestOptimizeRepository_ConcurrencyLimit(t *testing.T) { repo := localrepo.NewTestRepo(t, cfg, repoProto) manager := NewManager(gitalycfgprom.Config{}, nil) - manager.optimizeFunc = func(context.Context, *RepositoryManager, *localrepo.Repo, OptimizationStrategy) error { + manager.optimizeFunc = func(context.Context, *RepositoryManager, logrus.FieldLogger, *localrepo.Repo, OptimizationStrategy) error { // This should only happen if housekeeping is running successfully. // So by sending data on this channel we can notify the test that this // function ran successfully. @@ -1126,7 +1127,7 @@ func TestOptimizeRepository_ConcurrencyLimit(t *testing.T) { } go func() { - require.NoError(t, manager.OptimizeRepository(ctx, repo)) + require.NoError(t, manager.OptimizeRepository(ctx, testhelper.SharedLogger(t), repo)) }() // Only if optimizeFunc is run, we shall receive data here, this acts as test that @@ -1141,7 +1142,7 @@ func TestOptimizeRepository_ConcurrencyLimit(t *testing.T) { repo := localrepo.NewTestRepo(t, cfg, repoProto) manager := NewManager(gitalycfgprom.Config{}, nil) - manager.optimizeFunc = func(context.Context, *RepositoryManager, *localrepo.Repo, OptimizationStrategy) error { + manager.optimizeFunc = func(context.Context, *RepositoryManager, logrus.FieldLogger, *localrepo.Repo, OptimizationStrategy) error { require.FailNow(t, "housekeeping run should have been skipped") return nil } @@ -1152,7 +1153,7 @@ func TestOptimizeRepository_ConcurrencyLimit(t *testing.T) { // check that the state actually exists. require.Contains(t, manager.repositoryStates.values, repoPath) - require.NoError(t, manager.OptimizeRepository(ctx, repo)) + require.NoError(t, manager.OptimizeRepository(ctx, testhelper.SharedLogger(t), repo)) // After running the cleanup, the state should be removed. cleanup() @@ -1174,7 +1175,7 @@ func TestOptimizeRepository_ConcurrencyLimit(t *testing.T) { reposOptimized := make(map[string]struct{}) manager := NewManager(gitalycfgprom.Config{}, nil) - manager.optimizeFunc = func(_ context.Context, _ *RepositoryManager, repo *localrepo.Repo, _ OptimizationStrategy) error { + manager.optimizeFunc = func(_ context.Context, _ *RepositoryManager, _ logrus.FieldLogger, repo *localrepo.Repo, _ OptimizationStrategy) error { reposOptimized[repo.GetRelativePath()] = struct{}{} if repo.GetRelativePath() == repoFirst.GetRelativePath() { @@ -1188,13 +1189,13 @@ func TestOptimizeRepository_ConcurrencyLimit(t *testing.T) { // We block in the first call so that we can assert that a second call // to a different repository performs the optimization regardless without blocking. go func() { - require.NoError(t, manager.OptimizeRepository(ctx, repoFirst)) + require.NoError(t, manager.OptimizeRepository(ctx, testhelper.SharedLogger(t), repoFirst)) }() <-reqReceivedCh // Because this optimizes a different repository this call shouldn't block. - require.NoError(t, manager.OptimizeRepository(ctx, repoSecond)) + require.NoError(t, manager.OptimizeRepository(ctx, testhelper.SharedLogger(t), repoSecond)) <-ch @@ -1211,7 +1212,7 @@ func TestOptimizeRepository_ConcurrencyLimit(t *testing.T) { var optimizations int manager := NewManager(gitalycfgprom.Config{}, nil) - manager.optimizeFunc = func(context.Context, *RepositoryManager, *localrepo.Repo, OptimizationStrategy) error { + manager.optimizeFunc = func(context.Context, *RepositoryManager, logrus.FieldLogger, *localrepo.Repo, OptimizationStrategy) error { optimizations++ if optimizations == 1 { @@ -1226,7 +1227,7 @@ func TestOptimizeRepository_ConcurrencyLimit(t *testing.T) { wg.Add(1) go func() { defer wg.Done() - require.NoError(t, manager.OptimizeRepository(ctx, repo)) + require.NoError(t, manager.OptimizeRepository(ctx, testhelper.SharedLogger(t), repo)) }() <-reqReceivedCh @@ -1235,9 +1236,9 @@ func TestOptimizeRepository_ConcurrencyLimit(t *testing.T) { // that all subsequent calls which try to optimize the same repository return immediately. // Furthermore, we expect to see only a single call to the optimizing function because we // don't want to optimize the same repository concurrently. - require.NoError(t, manager.OptimizeRepository(ctx, repo)) - require.NoError(t, manager.OptimizeRepository(ctx, repo)) - require.NoError(t, manager.OptimizeRepository(ctx, repo)) + require.NoError(t, manager.OptimizeRepository(ctx, testhelper.SharedLogger(t), repo)) + require.NoError(t, manager.OptimizeRepository(ctx, testhelper.SharedLogger(t), repo)) + require.NoError(t, manager.OptimizeRepository(ctx, testhelper.SharedLogger(t), repo)) assert.Equal(t, 1, optimizations) <-ch @@ -1246,9 +1247,9 @@ func TestOptimizeRepository_ConcurrencyLimit(t *testing.T) { // When performing optimizations sequentially though the repository // should be unlocked after every call, and consequentially we should // also see multiple calls to the optimizing function. - require.NoError(t, manager.OptimizeRepository(ctx, repo)) - require.NoError(t, manager.OptimizeRepository(ctx, repo)) - require.NoError(t, manager.OptimizeRepository(ctx, repo)) + require.NoError(t, manager.OptimizeRepository(ctx, testhelper.SharedLogger(t), repo)) + require.NoError(t, manager.OptimizeRepository(ctx, testhelper.SharedLogger(t), repo)) + require.NoError(t, manager.OptimizeRepository(ctx, testhelper.SharedLogger(t), repo)) assert.Equal(t, 4, optimizations) }) } diff --git a/internal/git/objectpool/fetch.go b/internal/git/objectpool/fetch.go index f2a9b10b7..82f452b39 100644 --- a/internal/git/objectpool/fetch.go +++ b/internal/git/objectpool/fetch.go @@ -35,7 +35,7 @@ func (o *ObjectPool) FetchFromOrigin(ctx context.Context, origin *localrepo.Repo return fmt.Errorf("computing origin repo's path: %w", err) } - if err := o.housekeepingManager.CleanStaleData(ctx, o.Repo, housekeeping.DefaultStaleDataCleanup()); err != nil { + if err := o.housekeepingManager.CleanStaleData(ctx, logger, o.Repo, housekeeping.DefaultStaleDataCleanup()); err != nil { return fmt.Errorf("cleaning stale data: %w", err) } @@ -105,7 +105,7 @@ func (o *ObjectPool) FetchFromOrigin(ctx context.Context, origin *localrepo.Repo return fmt.Errorf("computing stats after fetch: %w", err) } - if err := o.housekeepingManager.OptimizeRepository(ctx, o.Repo); err != nil { + if err := o.housekeepingManager.OptimizeRepository(ctx, logger, o.Repo); err != nil { return fmt.Errorf("optimizing pool repo: %w", err) } diff --git a/internal/gitaly/maintenance/optimize.go b/internal/gitaly/maintenance/optimize.go index 3369ba1ff..6233b9aa4 100644 --- a/internal/gitaly/maintenance/optimize.go +++ b/internal/gitaly/maintenance/optimize.go @@ -9,7 +9,6 @@ import ( "path/filepath" "time" - "github.com/grpc-ecosystem/go-grpc-middleware/logging/logrus/ctxlogrus" "github.com/sirupsen/logrus" "gitlab.com/gitlab-org/gitaly/v16/internal/gitaly/config" "gitlab.com/gitlab-org/gitaly/v16/internal/gitaly/storage" @@ -68,15 +67,15 @@ func shuffledStoragesCopy(randSrc *rand.Rand, storages []config.Storage) []confi // Optimizer knows how to optimize a repository type Optimizer interface { - OptimizeRepository(context.Context, storage.Repository) error + OptimizeRepository(context.Context, logrus.FieldLogger, storage.Repository) error } // OptimizerFunc is an adapter to allow the use of an ordinary function as an Optimizer -type OptimizerFunc func(context.Context, storage.Repository) error +type OptimizerFunc func(context.Context, logrus.FieldLogger, storage.Repository) error // OptimizeRepository calls o(ctx, repo) -func (o OptimizerFunc) OptimizeRepository(ctx context.Context, repo storage.Repository) error { - return o(ctx, repo) +func (o OptimizerFunc) OptimizeRepository(ctx context.Context, logger logrus.FieldLogger, repo storage.Repository) error { + return o(ctx, logger, repo) } // DailyOptimizationWorker creates a worker that runs repository maintenance daily @@ -110,9 +109,7 @@ func optimizeRepo( "start_time": start.UTC(), }) - ctx = ctxlogrus.ToContext(ctx, logEntry) - - err := o.OptimizeRepository(ctx, repo) + err := o.OptimizeRepository(ctx, l, repo) logEntry = logEntry.WithField("time_ms", time.Since(start).Milliseconds()) if err != nil { diff --git a/internal/gitaly/maintenance/optimize_test.go b/internal/gitaly/maintenance/optimize_test.go index 5b94b2f98..7ead8cd01 100644 --- a/internal/gitaly/maintenance/optimize_test.go +++ b/internal/gitaly/maintenance/optimize_test.go @@ -5,6 +5,7 @@ import ( "math/rand" "testing" + "github.com/sirupsen/logrus" "github.com/stretchr/testify/require" "gitlab.com/gitlab-org/gitaly/v16/internal/git/catfile" "gitlab.com/gitlab-org/gitaly/v16/internal/git/gittest" @@ -26,7 +27,7 @@ type mockOptimizer struct { cfg config.Cfg } -func (mo *mockOptimizer) OptimizeRepository(ctx context.Context, repository storage.Repository) error { +func (mo *mockOptimizer) OptimizeRepository(ctx context.Context, logger logrus.FieldLogger, repository storage.Repository) error { mo.actual = append(mo.actual, repository) l := config.NewLocator(mo.cfg) gitCmdFactory := gittest.NewCommandFactory(mo.t, mo.cfg) @@ -35,7 +36,7 @@ func (mo *mockOptimizer) OptimizeRepository(ctx context.Context, repository stor txManager := transaction.NewManager(mo.cfg, backchannel.NewRegistry()) housekeepingManager := housekeeping.NewManager(mo.cfg.Prometheus, txManager) - return housekeepingManager.OptimizeRepository(ctx, localrepo.New(l, gitCmdFactory, catfileCache, repository)) + return housekeepingManager.OptimizeRepository(ctx, logger, localrepo.New(l, gitCmdFactory, catfileCache, repository)) } func TestOptimizeReposRandomly(t *testing.T) { @@ -103,13 +104,15 @@ func TestOptimizeReposRandomly(t *testing.T) { tickerDone = true } + logger := testhelper.SharedLogger(t) + mo := &mockOptimizer{ t: t, cfg: cfg, } walker := OptimizeReposRandomly(cfg, mo, ticker, rand.New(rand.NewSource(1))) - require.NoError(t, walker(ctx, testhelper.SharedLogger(t), tc.storages)) + require.NoError(t, walker(ctx, logger, tc.storages)) require.ElementsMatch(t, tc.expected, mo.actual) require.True(t, tickerDone) // We expect one more tick than optimized repositories because of the diff --git a/internal/gitaly/service/ref/find_local_branches_test.go b/internal/gitaly/service/ref/find_local_branches_test.go index 3e3064901..e47b82214 100644 --- a/internal/gitaly/service/ref/find_local_branches_test.go +++ b/internal/gitaly/service/ref/find_local_branches_test.go @@ -9,6 +9,7 @@ import ( "github.com/stretchr/testify/require" "gitlab.com/gitlab-org/gitaly/v16/internal/git" "gitlab.com/gitlab-org/gitaly/v16/internal/git/gittest" + "gitlab.com/gitlab-org/gitaly/v16/internal/git/updateref" "gitlab.com/gitlab-org/gitaly/v16/internal/gitaly/storage" "gitlab.com/gitlab-org/gitaly/v16/internal/structerr" "gitlab.com/gitlab-org/gitaly/v16/internal/testhelper" @@ -233,6 +234,42 @@ func TestFindLocalBranches(t *testing.T) { }, }, { + desc: "with pagination limit and many filtered branches", + setup: func(t *testing.T) setupData { + repo, repoPath := gittest.CreateRepository(t, ctx, cfg) + + commitID, commit := writeCommit(t, ctx, cfg, repo, gittest.WithBranch("branch-a"), gittest.WithMessage("commit a")) + + updater, err := updateref.New(ctx, gittest.NewRepositoryPathExecutor(t, cfg, repoPath)) + require.NoError(t, err) + + // Create a bunch of local branches. When we only ask for a very small subset, this will + // cause us to `Wait()` for git-for-each-ref(1) early while it is still enumerating the + // branches. We thus close its stdout early and then wait for it to terminate, which can + // lead to the process receiving the EPIPE signal if we didn't take proper care. + // + // So the more references we have, the more likely it is that we'll see the issue. + require.NoError(t, updater.Start()) + for i := 0; i < 1000; i++ { + require.NoError(t, updater.Create(git.ReferenceName(fmt.Sprintf("refs/heads/branch-%d", i)), commitID)) + } + require.NoError(t, updater.Commit()) + + return setupData{ + request: &gitalypb.FindLocalBranchesRequest{ + Repository: repo, + PaginationParams: &gitalypb.PaginationParameter{ + Limit: 2, + }, + }, + expectedBranches: []*gitalypb.Branch{ + {Name: []byte("refs/heads/branch-0"), TargetCommit: commit}, + {Name: []byte("refs/heads/branch-1"), TargetCommit: commit}, + }, + } + }, + }, + { desc: "invalid page token", setup: func(t *testing.T) setupData { repo, _ := gittest.CreateRepository(t, ctx, cfg) @@ -246,7 +283,7 @@ func TestFindLocalBranches(t *testing.T) { PageToken: "refs/heads/does-not-exist", }, }, - expectedErr: structerr.NewInternal("finding refs: could not find page token"), + expectedErr: structerr.NewInternal("finding refs: sending lines: could not find page token"), } }, }, diff --git a/internal/gitaly/service/ref/util.go b/internal/gitaly/service/ref/util.go index 5fb74f82d..4f90c3b0d 100644 --- a/internal/gitaly/service/ref/util.go +++ b/internal/gitaly/service/ref/util.go @@ -3,12 +3,17 @@ package ref import ( "bytes" "context" + "errors" "fmt" "math" + "os/exec" + "strings" + "syscall" "gitlab.com/gitlab-org/gitaly/v16/internal/git" "gitlab.com/gitlab-org/gitaly/v16/internal/git/catfile" "gitlab.com/gitlab-org/gitaly/v16/internal/helper/lines" + "gitlab.com/gitlab-org/gitaly/v16/internal/structerr" "gitlab.com/gitlab-org/gitaly/v16/proto/go/gitalypb" ) @@ -129,13 +134,14 @@ func (s *server) findRefs(ctx context.Context, writer lines.Sender, repo git.Rep options = append(options, opts.cmdArgs...) } + var stderr strings.Builder cmd, err := repo.Exec(ctx, git.Command{ Name: "for-each-ref", Flags: options, Args: patterns, - }, git.WithSetupStdout()) + }, git.WithSetupStdout(), git.WithStderr(&stderr)) if err != nil { - return err + return fmt.Errorf("spawning for-each-ref: %w", err) } if err := lines.Send(cmd, writer, lines.SenderOpts{ @@ -144,10 +150,31 @@ func (s *server) findRefs(ctx context.Context, writer lines.Sender, repo git.Rep Limit: opts.Limit, PageTokenError: opts.PageTokenError, }); err != nil { - return err + return fmt.Errorf("sending lines: %w", err) } - return cmd.Wait() + if err := cmd.Wait(); err != nil { + var exitErr *exec.ExitError + if errors.As(err, &exitErr) { + // When we have a limit set up and have sent all references upstream then the call to `Wait()` may + // indeed cause us to tear down the still-running git-for-each-ref(1) process. Because we close stdout + // before sending a signal the end result may be that the process will die with EPIPE because it failed + // to write to stdout. + // + // This is an expected error though, and thus we ignore it here. + status, ok := exitErr.ProcessState.Sys().(syscall.WaitStatus) + if ok && status.Signaled() && status.Signal() == syscall.SIGPIPE { + return nil + } + + return structerr.New("listing failed with exit code %d", status.ExitStatus()). + WithMetadata("stderr", stderr.String()) + } + + return fmt.Errorf("waiting for for-each-ref: %w", err) + } + + return nil } type paginationOpts struct { diff --git a/internal/gitaly/service/repository/optimize.go b/internal/gitaly/service/repository/optimize.go index 02c8b9fa2..9fb563d0d 100644 --- a/internal/gitaly/service/repository/optimize.go +++ b/internal/gitaly/service/repository/optimize.go @@ -4,6 +4,7 @@ import ( "context" "fmt" + "github.com/grpc-ecosystem/go-grpc-middleware/logging/logrus/ctxlogrus" "gitlab.com/gitlab-org/gitaly/v16/internal/git/housekeeping" "gitlab.com/gitlab-org/gitaly/v16/internal/git/stats" "gitlab.com/gitlab-org/gitaly/v16/internal/structerr" @@ -36,7 +37,7 @@ func (s *server) OptimizeRepository(ctx context.Context, in *gitalypb.OptimizeRe return nil, structerr.NewInvalidArgument("unsupported optimization strategy %d", in.GetStrategy()) } - if err := s.housekeepingManager.OptimizeRepository(ctx, repo, + if err := s.housekeepingManager.OptimizeRepository(ctx, ctxlogrus.Extract(ctx), repo, housekeeping.WithOptimizationStrategyConstructor(strategyConstructor), ); err != nil { return nil, structerr.NewInternal("%w", err) diff --git a/internal/gitaly/service/repository/optimize_test.go b/internal/gitaly/service/repository/optimize_test.go index 97e72da6a..c05086c23 100644 --- a/internal/gitaly/service/repository/optimize_test.go +++ b/internal/gitaly/service/repository/optimize_test.go @@ -10,6 +10,7 @@ import ( "testing" "time" + "github.com/sirupsen/logrus" "github.com/sirupsen/logrus/hooks/test" "github.com/stretchr/testify/require" "gitlab.com/gitlab-org/gitaly/v16/internal/git/gittest" @@ -258,7 +259,7 @@ type mockHousekeepingManager struct { strategyCh chan housekeeping.OptimizationStrategy } -func (m mockHousekeepingManager) OptimizeRepository(_ context.Context, _ *localrepo.Repo, opts ...housekeeping.OptimizeRepositoryOption) error { +func (m mockHousekeepingManager) OptimizeRepository(_ context.Context, _ logrus.FieldLogger, _ *localrepo.Repo, opts ...housekeeping.OptimizeRepositoryOption) error { var cfg housekeeping.OptimizeRepositoryConfig for _, opt := range opts { opt(&cfg) diff --git a/internal/gitaly/storage/storagemgr/transaction_manager.go b/internal/gitaly/storage/storagemgr/transaction_manager.go index 6ba849132..d95d9fd78 100644 --- a/internal/gitaly/storage/storagemgr/transaction_manager.go +++ b/internal/gitaly/storage/storagemgr/transaction_manager.go @@ -1369,7 +1369,7 @@ func (mgr *TransactionManager) prepareReferenceTransaction(ctx context.Context, // We ask housekeeping to cleanup stale reference locks. We don't add a grace period, because // transaction manager is the only process which writes into the repository, so it is safe // to delete these locks. - if err := mgr.housekeepingManager.CleanStaleData(ctx, mgr.repository, housekeeping.OnlyStaleReferenceLockCleanup(0)); err != nil { + if err := mgr.housekeepingManager.CleanStaleData(ctx, ctxlogrus.Extract(ctx), mgr.repository, housekeeping.OnlyStaleReferenceLockCleanup(0)); err != nil { return nil, fmt.Errorf("running reflock cleanup: %w", err) } |