<commit_before>package api

import (
	"context"
	"fmt"
	"net/http"
	"strconv"
	"time"

	"github.com/go-gorp/gorp"
	"github.com/golang/protobuf/ptypes"
	"github.com/ovh/venom"
	"github.com/sguiheux/go-coverage"

	"github.com/ovh/cds/engine/api/cache"
	"github.com/ovh/cds/engine/api/event"
	"github.com/ovh/cds/engine/api/group"
	"github.com/ovh/cds/engine/api/metrics"
	"github.com/ovh/cds/engine/api/observability"
	"github.com/ovh/cds/engine/api/permission"
	"github.com/ovh/cds/engine/api/project"
	"github.com/ovh/cds/engine/api/repositoriesmanager"
	"github.com/ovh/cds/engine/api/worker"
	"github.com/ovh/cds/engine/api/workermodel"
	"github.com/ovh/cds/engine/api/workflow"
	"github.com/ovh/cds/engine/service"
	"github.com/ovh/cds/sdk"
	"github.com/ovh/cds/sdk/cdsclient"
	"github.com/ovh/cds/sdk/log"
)

func (api *API) postTakeWorkflowJobHandler() service.Handler {
	return func(ctx context.Context, w http.ResponseWriter, r *http.Request) error {
		id, errc := requestVarInt(r, "id")
		if errc != nil {
			return sdk.WrapError(errc, "Invalid id")
		}

		takeForm := &sdk.WorkerTakeForm{}
		if err := service.UnmarshalBody(r, takeForm); err != nil {
			return sdk.WrapError(err, "Cannot unmarshal request")
		}

		user := deprecatedGetUser(ctx)
		p, errP := project.LoadProjectByNodeJobRunID(ctx, api.mustDB(), api.Cache, id, user, project.LoadOptions.WithVariables, project.LoadOptions.WithClearKeys)
		if errP != nil {
			var username string
			if user != nil {
				username = user.Username
			}
			return sdk.WrapError(errP, "Cannot load project by nodeJobRunID:%d requester:%s", id, username)
		}

		//Load worker model
		workerModel := getWorker(ctx).Name
		if getWorker(ctx).ModelID != 0 {
			wm, errModel := workermodel.LoadByID(api.mustDB(), getWorker(ctx).ModelID)
			if errModel != nil {
				return sdk.ErrNoWorkerModel
			}
			workerModel = wm.Name
		}

		pbj, errl := workflow.LoadNodeJobRun(api.mustDB(), api.Cache, id)
		if errl != nil {
			return sdk.WrapError(errl, "Cannot load job nodeJobRunID:%d", id)
		}

		observability.Current(ctx,
			observability.Tag(observability.TagWorkflowNodeJobRun, id),
			observability.Tag(observability.TagWorkflowNodeRun, pbj.WorkflowNodeRunID),
			observability.Tag(observability.TagJob, pbj.Job.Action.Name))

		// a worker can have only one group
		groups := deprecatedGetUser(ctx).Groups
		if len(groups) != 1 {
			return sdk.WrapError(errl, "Too many groups detected on worker:%d", len(groups))
		}

		var isGroupOK bool
		if len(pbj.ExecGroups) == 0 {
			isGroupOK = true
		} else {
			for _, g := range pbj.ExecGroups {
				if g.ID == groups[0].ID {
					isGroupOK = true
					break
				}
			}
		}

		if !isGroupOK {
			return sdk.WrapError(sdk.ErrForbidden, "This worker is not authorized to take this job:%d execGroups:%+v", id, pbj.ExecGroups)
		}

		pbji := &sdk.WorkflowNodeJobRunData{}
		report, errT := takeJob(ctx, api.mustDB, api.Cache, p, id, takeForm, workerModel, pbji)
		if errT != nil {
			return sdk.WrapError(errT, "Cannot takeJob nodeJobRunID:%d", id)
		}

		workflow.ResyncNodeRunsWithCommits(api.mustDB(), api.Cache, p, report)
		go workflow.SendEvent(api.mustDB(), p.Key, report)

		return service.WriteJSON(w, pbji, http.StatusOK)
	}
}

func takeJob(ctx context.Context, dbFunc func() *gorp.DbMap, store cache.Store, p *sdk.Project, id int64, takeForm *sdk.WorkerTakeForm, workerModel string, wnjri *sdk.WorkflowNodeJobRunData) (*workflow.ProcessorReport, error) {
	// Start a tx
	tx, errBegin := dbFunc().Begin()
	if errBegin != nil {
		return nil, sdk.WrapError(errBegin, "Cannot start transaction")
	}
	defer tx.Rollback()

	//Prepare spawn infos
	infos := []sdk.SpawnInfo{
		{
			RemoteTime: takeForm.Time,<commit_msg>fix(api): improve logs on postTakeWorkflowJobHandler errors (#4385)<commit_after>package api

import (
	"context"
	"fmt"
	"net/http"
	"strconv"
	"time"

	"github.com/go-gorp/gorp"
	"github.com/golang/protobuf/ptypes"
	"github.com/ovh/venom"
	"github.com/sguiheux/go-coverage"

	"github.com/ovh/cds/engine/api/cache"
	"github.com/ovh/cds/engine/api/event"
	"github.com/ovh/cds/engine/api/group"
	"github.com/ovh/cds/engine/api/metrics"
	"github.com/ovh/cds/engine/api/observability"
	"github.com/ovh/cds/engine/api/permission"
	"github.com/ovh/cds/engine/api/project"
	"github.com/ovh/cds/engine/api/repositoriesmanager"
	"github.com/ovh/cds/engine/api/worker"
	"github.com/ovh/cds/engine/api/workermodel"
	"github.com/ovh/cds/engine/api/workflow"
	"github.com/ovh/cds/engine/service"
	"github.com/ovh/cds/sdk"
	"github.com/ovh/cds/sdk/cdsclient"
	"github.com/ovh/cds/sdk/log"
)

func (api *API) postTakeWorkflowJobHandler() service.Handler {
	return func(ctx context.Context, w http.ResponseWriter, r *http.Request) error {
		id, errc := requestVarInt(r, "id")
		if errc != nil {
			return sdk.WrapError(errc, "Invalid id")
		}

		takeForm := &sdk.WorkerTakeForm{}
		if err := service.UnmarshalBody(r, takeForm); err != nil {
			return sdk.WrapError(err, "Cannot unmarshal request")
		}

		user := deprecatedGetUser(ctx)
		p, errP := project.LoadProjectByNodeJobRunID(ctx, api.mustDB(), api.Cache, id, user, project.LoadOptions.WithVariables, project.LoadOptions.WithClearKeys)
		if errP != nil {
			var username string
			if user != nil {
				username = user.Username
			}
			return sdk.WrapError(errP, "Cannot load project by nodeJobRunID:%d requester:%s", id, username)
		}

		//Load worker model
		workerModel := getWorker(ctx).Name
		if getWorker(ctx).ModelID != 0 {
			wm, errModel := workermodel.LoadByID(api.mustDB(), getWorker(ctx).ModelID)
			if errModel != nil {
				return sdk.ErrNoWorkerModel
			}
			workerModel = wm.Name
		}

		pbj, errl := workflow.LoadNodeJobRun(api.mustDB(), api.Cache, id)
		if errl != nil {
			return sdk.WrapError(errl, "Cannot load job nodeJobRunID:%d", id)
		}

		observability.Current(ctx,
			observability.Tag(observability.TagWorkflowNodeJobRun, id),
			observability.Tag(observability.TagWorkflowNodeRun, pbj.WorkflowNodeRunID),
			observability.Tag(observability.TagJob, pbj.Job.Action.Name))

		// a worker can have only one group
		groups := deprecatedGetUser(ctx).Groups
		if len(groups) != 1 {
			return sdk.WrapError(errl, "Too many groups detected on worker:%d", len(groups))
		}

		var isGroupOK bool
		if len(pbj.ExecGroups) == 0 {
			isGroupOK = true
		} else {
			for _, g := range pbj.ExecGroups {
				if g.ID == groups[0].ID {
					isGroupOK = true
					break
				}
			}
		}

		if !isGroupOK {
      return sdk.WrapError(sdk.ErrForbidden, "Worker %s (%s) is not authorized to take this job:%d execGroups:%+v", getWorker(ctx).Name, workerModel, id, pbj.ExecGroups)
		}

		pbji := &sdk.WorkflowNodeJobRunData{}
		report, errT := takeJob(ctx, api.mustDB, api.Cache, p, id, takeForm, workerModel, pbji)
		if errT != nil {
			return sdk.WrapError(errT, "Cannot takeJob nodeJobRunID:%d", id)
		}

		workflow.ResyncNodeRunsWithCommits(api.mustDB(), api.Cache, p, report)
		go workflow.SendEvent(api.mustDB(), p.Key, report)

		return service.WriteJSON(w, pbji, http.StatusOK)
	}
}

func takeJob(ctx context.Context, dbFunc func() *gorp.DbMap, store cache.Store, p *sdk.Project, id int64, takeForm *sdk.WorkerTakeForm, workerModel string, wnjri *sdk.WorkflowNodeJobRunData) (*workflow.ProcessorReport, error) {
	// Start a tx
	tx, errBegin := dbFunc().Begin()
	if errBegin != nil {
		return nil, sdk.WrapError(errBegin, "Cannot start transaction")
	}
	defer tx.Rollback()

	//Prepare spawn infos
	infos := []sdk.SpawnInfo{
		{
			RemoteTime: takeForm.Time,
{'fraction_non_alphanumeric': 0.14329430548163916, 'fraction_numerical': 0.0021287919105907396, 'mean_word_length': 4.072199730094467, 'pattern_counts': {'":': 0, '<': 3, '<?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': '5741307', 'n_tokens_mistral': 3051, 'n_tokens_neox': 2802, 'n_words': 558}