<commit_before>// Copyright 2020 The LUCI Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//      http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package impl

import (
	"context"
	"fmt"
	"strings"
	"time"

	"google.golang.org/protobuf/encoding/protojson"
	"google.golang.org/protobuf/proto"

	"go.chromium.org/luci/common/clock"
	"go.chromium.org/luci/common/errors"
	"go.chromium.org/luci/common/logging"
	"go.chromium.org/luci/common/retry/transient"
	"go.chromium.org/luci/gae/service/datastore"
	"go.chromium.org/luci/server/tq"

	"go.chromium.org/luci/cv/internal/buildbucket"
	"go.chromium.org/luci/cv/internal/changelist"
	"go.chromium.org/luci/cv/internal/common"
	"go.chromium.org/luci/cv/internal/common/bq"
	"go.chromium.org/luci/cv/internal/common/eventbox"
	"go.chromium.org/luci/cv/internal/common/lease"
	"go.chromium.org/luci/cv/internal/common/tree"
	"go.chromium.org/luci/cv/internal/gerrit"
	"go.chromium.org/luci/cv/internal/gerrit/cancel"
	"go.chromium.org/luci/cv/internal/prjmanager"
	"go.chromium.org/luci/cv/internal/run"
	runbq "go.chromium.org/luci/cv/internal/run/bq"
	"go.chromium.org/luci/cv/internal/run/eventpb"
	"go.chromium.org/luci/cv/internal/run/impl/handler"
	"go.chromium.org/luci/cv/internal/run/impl/longops"
	"go.chromium.org/luci/cv/internal/run/impl/state"
	"go.chromium.org/luci/cv/internal/run/impl/submit"
	"go.chromium.org/luci/cv/internal/run/pubsub"
)

// maxEventsPerBatch limits the number of incoming events the PM will process at
// once.
//
// This shouldn't be hit in practice under normal operation. This is chosen such
// that RM can read these events and make some progress in 1 minute.
const maxEventsPerBatch = 10000

// RunManager manages Runs.
//
// It decides starting, cancellation, submission etc. for Runs.
type RunManager struct {
	runNotifier  *run.Notifier
	pmNotifier   *prjmanager.Notifier
	tqDispatcher *tq.Dispatcher
	env          *common.Env
	gFactory     gerrit.Factory
	bbFactory    buildbucket.ClientFactory
	handler      handler.Handler

	testDoLongOperationWithDeadline func(context.Context, *longops.Base) (*eventpb.LongOpCompleted, error)
}

// New constructs a new RunManager instance.
func New(
	n *run.Notifier,
	pm *prjmanager.Notifier,
	clm *changelist.Mutator,
	clu *changelist.Updater,
	g gerrit.Factory,
	bb buildbucket.ClientFactory,
	tc tree.Client,
	bqc bq.Client,
	env *common.Env,
) *RunManager {
	rm := &RunManager{
		runNotifier:  n,
		pmNotifier:   pm,
		tqDispatcher: n.TasksBinding.TQDispatcher,
		env:          env,
		gFactory:     g,
		bbFactory:    bb,
		handler: &handler.Impl{
			PM:         pm,
			RM:         n,
			CLUpdater:  clu,
			CLMutator:  clm,
			BQExporter: runbq.NewExporter(n.TasksBinding.TQDispatcher, bqc, env),
			GFactory:   g,
			TreeClient: tc,
			Publisher:  pubsub.NewPublisher(n.TasksBinding.TQDispatcher, env),
		},
	}
	n.TasksBinding.ManageRun.AttachHandler(
		func(ctx context.Context, payload proto.Message) error {
			task := payload.(*eventpb.ManageRunTask)
			ctx = logging.SetField(ctx, "run", task.GetRunId())
			err := rm.manageRun(ctx, common.RunID(task.GetRunId()))<commit_msg>cv: pass env when initializing handler<commit_after>// Copyright 2020 The LUCI Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//      http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package impl

import (
	"context"
	"fmt"
	"strings"
	"time"

	"google.golang.org/protobuf/encoding/protojson"
	"google.golang.org/protobuf/proto"

	"go.chromium.org/luci/common/clock"
	"go.chromium.org/luci/common/errors"
	"go.chromium.org/luci/common/logging"
	"go.chromium.org/luci/common/retry/transient"
	"go.chromium.org/luci/gae/service/datastore"
	"go.chromium.org/luci/server/tq"

	"go.chromium.org/luci/cv/internal/buildbucket"
	"go.chromium.org/luci/cv/internal/changelist"
	"go.chromium.org/luci/cv/internal/common"
	"go.chromium.org/luci/cv/internal/common/bq"
	"go.chromium.org/luci/cv/internal/common/eventbox"
	"go.chromium.org/luci/cv/internal/common/lease"
	"go.chromium.org/luci/cv/internal/common/tree"
	"go.chromium.org/luci/cv/internal/gerrit"
	"go.chromium.org/luci/cv/internal/gerrit/cancel"
	"go.chromium.org/luci/cv/internal/prjmanager"
	"go.chromium.org/luci/cv/internal/run"
	runbq "go.chromium.org/luci/cv/internal/run/bq"
	"go.chromium.org/luci/cv/internal/run/eventpb"
	"go.chromium.org/luci/cv/internal/run/impl/handler"
	"go.chromium.org/luci/cv/internal/run/impl/longops"
	"go.chromium.org/luci/cv/internal/run/impl/state"
	"go.chromium.org/luci/cv/internal/run/impl/submit"
	"go.chromium.org/luci/cv/internal/run/pubsub"
)

// maxEventsPerBatch limits the number of incoming events the PM will process at
// once.
//
// This shouldn't be hit in practice under normal operation. This is chosen such
// that RM can read these events and make some progress in 1 minute.
const maxEventsPerBatch = 10000

// RunManager manages Runs.
//
// It decides starting, cancellation, submission etc. for Runs.
type RunManager struct {
	runNotifier  *run.Notifier
	pmNotifier   *prjmanager.Notifier
	tqDispatcher *tq.Dispatcher
	env          *common.Env
	gFactory     gerrit.Factory
	bbFactory    buildbucket.ClientFactory
	handler      handler.Handler

	testDoLongOperationWithDeadline func(context.Context, *longops.Base) (*eventpb.LongOpCompleted, error)
}

// New constructs a new RunManager instance.
func New(
	n *run.Notifier,
	pm *prjmanager.Notifier,
	clm *changelist.Mutator,
	clu *changelist.Updater,
	g gerrit.Factory,
	bb buildbucket.ClientFactory,
	tc tree.Client,
	bqc bq.Client,
	env *common.Env,
) *RunManager {
	rm := &RunManager{
		runNotifier:  n,
		pmNotifier:   pm,
		tqDispatcher: n.TasksBinding.TQDispatcher,
		env:          env,
		gFactory:     g,
		bbFactory:    bb,
		handler: &handler.Impl{
			PM:         pm,
			RM:         n,
			CLUpdater:  clu,
			CLMutator:  clm,
			BQExporter: runbq.NewExporter(n.TasksBinding.TQDispatcher, bqc, env),
			GFactory:   g,
			TreeClient: tc,
			Publisher:  pubsub.NewPublisher(n.TasksBinding.TQDispatcher, env),
			Env:        env,
		},
	}
	n.TasksBinding.ManageRun.AttachHandler(
		func(ctx context.Context, payload proto.Message) error {
			task := payload.(*eventpb.ManageRunTask)
			ctx = logging.SetField(ctx, "run", task.GetRunId())
			err := rm.manageRun(ctx, common.RunID(task.GetRunId()))
{'fraction_non_alphanumeric': 0.12664724649743375, 'fraction_numerical': 0.003884033846580663, 'mean_word_length': 3.7371879106438897, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '794701', 'n_tokens_mistral': 2767, 'n_tokens_neox': 2576, 'n_words': 546}