<commit_before>package resource

import (
	"context"
	"crypto/sha256"
	"fmt"
	"io"
	"time"

	"code.cloudfoundry.org/clock"
	"code.cloudfoundry.org/lager"
	"code.cloudfoundry.org/lager/lagerctx"
	"github.com/concourse/concourse/atc/db"
	"github.com/concourse/concourse/atc/db/lock"
	"github.com/concourse/concourse/atc/runtime"
)

const GetResourceLockInterval = 5 * time.Second

type Getter struct {
	lockFactory          lock.LockFactory
	clock                clock.Clock
	resourceCacheFactory db.ResourceCacheFactory
	volumeRepo           db.VolumeRepository
}

func NewGetter(lockFactory lock.LockFactory, clock clock.Clock, resourceCacheFactory db.ResourceCacheFactory, volumeRepo db.VolumeRepository) Getter {
	return Getter{
		lockFactory:          lockFactory,
		clock:                clock,
		resourceCacheFactory: resourceCacheFactory,
		volumeRepo:           volumeRepo,
	}
}

func (g Getter) Get(
	ctx context.Context,
	worker runtime.Worker,
	containerFactory func(ctx context.Context) (runtime.Container, []runtime.VolumeMount, error),
	getResource Resource,
	resourceCache db.UsedResourceCache,
	stderr io.Writer,
) (VersionResult, runtime.ProcessResult, runtime.Volume, error) {
	logger := lagerctx.FromContext(ctx)

	ticker := g.clock.NewTicker(GetResourceLockInterval)
	defer ticker.Stop()

	result, processResult, volume, ok, err := g.getUnderLock(ctx, logger, worker, containerFactory, getResource, resourceCache, stderr)
	if err != nil {
		return VersionResult{}, runtime.ProcessResult{}, nil, err
	}
	if ok {
		return result, processResult, volume, nil
	}

	for {
		select {
		case <-ctx.Done():
			return VersionResult{}, runtime.ProcessResult{}, nil, ctx.Err()

		case <-ticker.C():
			result, processResult, volume, ok, err := g.getUnderLock(ctx, logger, worker, containerFactory, getResource, resourceCache, stderr)
			if err != nil {
				return VersionResult{}, runtime.ProcessResult{}, nil, err
			}
			if ok {
				return result, processResult, volume, nil
			}
		}
	}
}

func (g Getter) getUnderLock(
	ctx context.Context,
	logger lager.Logger,
	worker runtime.Worker,
	containerFactory func(ctx context.Context) (runtime.Container, []runtime.VolumeMount, error),
	getResource Resource,
	resourceCache db.UsedResourceCache,
	stderr io.Writer,
) (VersionResult, runtime.ProcessResult, runtime.Volume, bool, error) {
	result, processResult, volume, found, err := g.findCache(ctx, logger, worker, resourceCache)
	if err != nil {
		return VersionResult{}, runtime.ProcessResult{}, nil, false, err
	}
	if found {
		return result, processResult, volume, true, nil
	}

	lockName := lockName(getResource, worker.Name())
	lockLogger := logger.Session("lock", lager.Data{"lock-name": lockName})
	lock, acquired, err := g.lockFactory.Acquire(lockLogger, lock.NewTaskLockID(lockName))
	if err != nil {
		lockLogger.Error("failed-to-get-lock", err)
		// not returning error for consistency with prior behaviour - we just
		// retry after GetResourceLockInterval
		return VersionResult{}, runtime.ProcessResult{}, nil, false, nil
	}

	if !acquired {
		lockLogger.Debug("did-not-get-lock")
		return VersionResult{}, runtime.ProcessResult{}, nil, false, nil
	}

	defer lock.Release()

	result, processResult, volume, err = g.getAndCreateCache(ctx, logger, worker, containerFactory, getResource, resourceCache, stderr)
	if err != nil {
		return VersionResult{}, runtime.ProcessResult{}, nil, false, err
	}

	return result, processResult, volume, true, nil
}

func (g Getter) findCache(
	ctx context.Context,
	logger lager.Logger,
	worker runtime.Worker,
	resourceCache db.UsedResourceCache,
) (VersionResult, runtime.ProcessResult, runtime.Volume, bool, error) {
	logger = logger.Session("find")

	dbVolume, found, err := g.volumeRepo.FindResourceCacheVolume(worker.Name(), resourceCache)
	if err != nil {
		logger.Error("failed-to-find-resource-cache-volume", err)
		return VersionResult{}, runtime.ProcessResult{}, nil, false, err
	}
	if !found {
		return VersionResult{}, runtime.ProcessResult{}, nil, false, nil
	}

	volume, found, err := worker.LookupVolume(logger, dbVolume.Handle())
	if err != nil {
		logger.Error("failed-to-lookup-resource-cache-volume-on-worker", err)
		return VersionResult{}, runtime.ProcessResult{}, nil, false, err
	}
	if !found {
		return VersionResult{}, runtime.ProcessResult{}, nil, false, nil
	}

	metadata, err := g.resourceCacheFactory.ResourceCacheMetadata(resourceCache)
	if err != nil {
		logger.Error("failed-to-get-resource-cache-metadata", err)
		return VersionResult{}, runtime.ProcessResult{}, nil, false, err
	}

	result := VersionResult{Version: resourceCache.Version(), Metadata: metadata.ToATCMetadata()}
	logger.Debug("found-initialized-versioned-source", lager.Data{
		"version":  result.Version,
		"metadata": result.Metadata,
	})

	return result, runtime.ProcessResult{ExitStatus: 0}, volume, true, nil
}

func (g Getter) getAndCreateCache(
	ctx context.Context,
	logger lager.Logger,
	worker runtime.Worker,
	containerFactory func(ctx context.Context) (runtime.Container, []runtime.VolumeMount, error),
	getResource Resource,
	resourceCache db.UsedResourceCache,
	stderr io.Writer,
) (VersionResult, runtime.ProcessResult, runtime.Volume, error) {
	result, processResult, volume, found, err := g.findCache(ctx, logger, worker, resourceCache)
	if err != nil {
		return VersionResult{}, runtime.ProcessResult{}, nil, err
	}
	if found {
		return result, processResult, volume, nil
	}

	logger = logger.Session("create")

	container, mounts, err := containerFactory(lagerctx.NewContext(ctx, logger))<commit_msg>B: stop logging version metadata in resource.Getter<commit_after>package resource

import (
	"context"
	"crypto/sha256"
	"fmt"
	"io"
	"time"

	"code.cloudfoundry.org/clock"
	"code.cloudfoundry.org/lager"
	"code.cloudfoundry.org/lager/lagerctx"
	"github.com/concourse/concourse/atc/db"
	"github.com/concourse/concourse/atc/db/lock"
	"github.com/concourse/concourse/atc/runtime"
)

const GetResourceLockInterval = 5 * time.Second

type Getter struct {
	lockFactory          lock.LockFactory
	clock                clock.Clock
	resourceCacheFactory db.ResourceCacheFactory
	volumeRepo           db.VolumeRepository
}

func NewGetter(lockFactory lock.LockFactory, clock clock.Clock, resourceCacheFactory db.ResourceCacheFactory, volumeRepo db.VolumeRepository) Getter {
	return Getter{
		lockFactory:          lockFactory,
		clock:                clock,
		resourceCacheFactory: resourceCacheFactory,
		volumeRepo:           volumeRepo,
	}
}

func (g Getter) Get(
	ctx context.Context,
	worker runtime.Worker,
	containerFactory func(ctx context.Context) (runtime.Container, []runtime.VolumeMount, error),
	getResource Resource,
	resourceCache db.UsedResourceCache,
	stderr io.Writer,
) (VersionResult, runtime.ProcessResult, runtime.Volume, error) {
	logger := lagerctx.FromContext(ctx)

	ticker := g.clock.NewTicker(GetResourceLockInterval)
	defer ticker.Stop()

	result, processResult, volume, ok, err := g.getUnderLock(ctx, logger, worker, containerFactory, getResource, resourceCache, stderr)
	if err != nil {
		return VersionResult{}, runtime.ProcessResult{}, nil, err
	}
	if ok {
		return result, processResult, volume, nil
	}

	for {
		select {
		case <-ctx.Done():
			return VersionResult{}, runtime.ProcessResult{}, nil, ctx.Err()

		case <-ticker.C():
			result, processResult, volume, ok, err := g.getUnderLock(ctx, logger, worker, containerFactory, getResource, resourceCache, stderr)
			if err != nil {
				return VersionResult{}, runtime.ProcessResult{}, nil, err
			}
			if ok {
				return result, processResult, volume, nil
			}
		}
	}
}

func (g Getter) getUnderLock(
	ctx context.Context,
	logger lager.Logger,
	worker runtime.Worker,
	containerFactory func(ctx context.Context) (runtime.Container, []runtime.VolumeMount, error),
	getResource Resource,
	resourceCache db.UsedResourceCache,
	stderr io.Writer,
) (VersionResult, runtime.ProcessResult, runtime.Volume, bool, error) {
	result, processResult, volume, found, err := g.findCache(ctx, logger, worker, resourceCache)
	if err != nil {
		return VersionResult{}, runtime.ProcessResult{}, nil, false, err
	}
	if found {
		return result, processResult, volume, true, nil
	}

	lockName := lockName(getResource, worker.Name())
	lockLogger := logger.Session("lock", lager.Data{"lock-name": lockName})
	lock, acquired, err := g.lockFactory.Acquire(lockLogger, lock.NewTaskLockID(lockName))
	if err != nil {
		lockLogger.Error("failed-to-get-lock", err)
		// not returning error for consistency with prior behaviour - we just
		// retry after GetResourceLockInterval
		return VersionResult{}, runtime.ProcessResult{}, nil, false, nil
	}

	if !acquired {
		lockLogger.Debug("did-not-get-lock")
		return VersionResult{}, runtime.ProcessResult{}, nil, false, nil
	}

	defer lock.Release()

	result, processResult, volume, err = g.getAndCreateCache(ctx, logger, worker, containerFactory, getResource, resourceCache, stderr)
	if err != nil {
		return VersionResult{}, runtime.ProcessResult{}, nil, false, err
	}

	return result, processResult, volume, true, nil
}

func (g Getter) findCache(
	ctx context.Context,
	logger lager.Logger,
	worker runtime.Worker,
	resourceCache db.UsedResourceCache,
) (VersionResult, runtime.ProcessResult, runtime.Volume, bool, error) {
	logger = logger.Session("find")

	dbVolume, found, err := g.volumeRepo.FindResourceCacheVolume(worker.Name(), resourceCache)
	if err != nil {
		logger.Error("failed-to-find-resource-cache-volume", err)
		return VersionResult{}, runtime.ProcessResult{}, nil, false, err
	}
	if !found {
		return VersionResult{}, runtime.ProcessResult{}, nil, false, nil
	}

	volume, found, err := worker.LookupVolume(logger, dbVolume.Handle())
	if err != nil {
		logger.Error("failed-to-lookup-resource-cache-volume-on-worker", err)
		return VersionResult{}, runtime.ProcessResult{}, nil, false, err
	}
	if !found {
		return VersionResult{}, runtime.ProcessResult{}, nil, false, nil
	}

	metadata, err := g.resourceCacheFactory.ResourceCacheMetadata(resourceCache)
	if err != nil {
		logger.Error("failed-to-get-resource-cache-metadata", err)
		return VersionResult{}, runtime.ProcessResult{}, nil, false, err
	}

	result := VersionResult{Version: resourceCache.Version(), Metadata: metadata.ToATCMetadata()}
	logger.Debug("found-initialized-versioned-source", lager.Data{"version": result.Version})

	return result, runtime.ProcessResult{ExitStatus: 0}, volume, true, nil
}

func (g Getter) getAndCreateCache(
	ctx context.Context,
	logger lager.Logger,
	worker runtime.Worker,
	containerFactory func(ctx context.Context) (runtime.Container, []runtime.VolumeMount, error),
	getResource Resource,
	resourceCache db.UsedResourceCache,
	stderr io.Writer,
) (VersionResult, runtime.ProcessResult, runtime.Volume, error) {
	result, processResult, volume, found, err := g.findCache(ctx, logger, worker, resourceCache)
	if err != nil {
		return VersionResult{}, runtime.ProcessResult{}, nil, err
	}
	if found {
		return result, processResult, volume, nil
	}

	logger = logger.Session("create")

	container, mounts, err := containerFactory(lagerctx.NewContext(ctx, logger))
{'fraction_non_alphanumeric': 0.12255517674502063, 'fraction_numerical': 0.0008971828458639871, 'mean_word_length': 4.907260201377849, 'pattern_counts': {'":': 5, '<': 7, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2448440', 'n_tokens_mistral': 3496, 'n_tokens_neox': 3364, 'n_words': 854}