// NewWorker returns a new apiserver-based raft transport worker,
// with the given configuration. The worker itself implements
// raft.Transport.
197203
func NewWorker(config Config) (worker.Worker, error) {
	if err := config.Validate(); err != nil {
		return nil, errors.Trace(err)
	}

	apiPorts := config.APIInfo.Ports()
	if n := len(apiPorts); n != 1 {
		return nil, errors.Errorf("api.Info has %d unique ports, expected 1", n)
	}

	w := &Worker{
		config:       config,
		connections:  make(chan net.Conn),
		dialRequests: make(chan dialRequest),
		apiPort:      apiPorts[0],
	}

	const logPrefix = "[transport] "
	logWriter := &raftutil.LoggoWriter{logger, loggo.DEBUG}
	logLogger := log.New(logWriter, logPrefix, 0)
	stream, err := newStreamLayer(config.LocalID, config.Hub, w.connections, config.Clock, &Dialer{
		APIInfo: config.APIInfo,
		DialRaw: w.dialRaw,
		Path:    config.Path,
	})
	if err != nil {
		return nil, errors.Trace(err)
	}
	transport := raft.NewNetworkTransportWithConfig(&raft.NetworkTransportConfig{
		Logger:  logLogger,
		MaxPool: maxPoolSize,
		Stream:  stream,
		Timeout: config.Timeout,
	})
	w.Transport = transport

	var h http.Handler = NewHandler(w.connections, w.catacomb.Dying())
	h = &httpcontext.BasicAuthHandler{
		Handler:       h,
		Authenticator: w.config.Authenticator,
		Authorizer:    httpcontext.AuthorizerFunc(controllerAuthorizer),
	}
	h = &httpcontext.ImpliedModelHandler{
		Handler:   h,
		ModelUUID: w.config.APIInfo.ModelTag.Id(),
	}

	w.config.Mux.AddHandler("GET", w.config.Path, h)

	if err := catacomb.Invoke(catacomb.Plan{
		Site: &w.catacomb,
		Work: func() error {
			defer transport.Close()
			defer w.config.Mux.RemoveHandler("GET", w.config.Path)
			return w.loop()
		},
		Init: []worker.Worker{stream},
	}); err != nil {
		transport.Close()
		w.config.Mux.RemoveHandler("GET", w.config.Path)
		return nil, errors.Trace(err)
	}
	return w, nil
}
1372864
func NewWorkerPool(ctx interface{}, concurrency uint, namespace string, pool *redis.Pool) *WorkerPool {
	return NewWorkerPoolWithOptions(ctx, concurrency, namespace, pool, WorkerPoolOptions{})
}
750248
func NewWorker(config FlagConfig) (worker.Worker, error) {
	worker, err := NewFlagWorker(config)
	if err != nil {
		return nil, errors.Trace(err)
	}
	return worker, nil
}
569025
func NewWorker(config Config) (worker.Worker, error) {
	if err := config.Validate(); err != nil {
		return nil, errors.Trace(err)
	}
	w := &forwarder{
		config: config,
	}
	unsubscribe, err := w.config.Hub.Subscribe(w.config.Topic, w.handleRequest)
	if err != nil {
		return nil, errors.Annotatef(err, "subscribing to %q", w.config.Topic)
	}
	w.unsubscribe = unsubscribe
	if err := catacomb.Invoke(catacomb.Plan{
		Site: &w.catacomb,
		Work: w.loop,
	}); err != nil {
		unsubscribe()
		return nil, errors.Trace(err)
	}
	return w, nil
}
2424
func NewWorker(config Config) (worker.Worker, error) {
	if err := config.Validate(); err != nil {
		return nil, errors.Trace(err)
	}

	mc, err := modelCredential(config.Facade)
	if err != nil {
		return nil, errors.Trace(err)
	}

	// This worker needs to monitor both the changes to the credential content that
	// this model uses as well as what credential the model uses.
	// It needs to be restarted if there is a change in either.
	mcw, err := config.Facade.WatchModelCredential()
	if err != nil {
		return nil, errors.Trace(err)
	}

	v := &validator{
		validatorFacade:        config.Facade,
		credential:             mc,
		modelCredentialWatcher: mcw,
	}

	// The watcher needs to be added to the worker's catacomb plan
	// here in order to be controlled by this worker's lifecycle events:
	// for example, to be destroyed when this worker is destroyed, etc.
	// We also add the watcher to the Plan.Init collection to ensure that
	// the worker's Plan.Work method is executed after the watcher
	// is initialised and watcher's changes collection obtains the changes.
	// Watchers that are added using catacomb.Add method
	// miss out on a first call of Worker's Plan.Work method and can, thus,
	// be missing out on an initial change.
	plan := catacomb.Plan{
		Site: &v.catacomb,
		Work: v.loop,
		Init: []worker.Worker{v.modelCredentialWatcher},
	}

	if mc.CloudCredential != "" {
		var err error
		v.credentialWatcher, err = config.Facade.WatchCredential(mc.CloudCredential)
		if err != nil {
			return nil, errors.Trace(err)
		}
		plan.Init = append(plan.Init, v.credentialWatcher)
	}

	if err := catacomb.Invoke(plan); err != nil {
		return nil, errors.Trace(err)
	}
	return v, nil
}
1058043
func newWorker(a agent.Agent) (worker.Worker, error) {
	metricsSpoolDir := a.CurrentConfig().MetricsSpoolDir()
	err := checkSpoolDir(metricsSpoolDir)
	if err != nil {
		return nil, errors.Annotatef(err, "error checking spool directory %q", metricsSpoolDir)
	}
	w := &spoolWorker{factory: newFactory(metricsSpoolDir)}
	w.tomb.Go(func() error {
		<-w.tomb.Dying()
		return nil
	})
	return w, nil
}
830329
func NewWorkerManager(id string, config *ConsumerConfig, topicPartition TopicAndPartition, metrics *ConsumerMetrics, closeConsumer chan bool) *WorkerManager {
	workers := make([]*Worker, config.NumWorkers)
	availableWorkers := make(chan *Worker, config.NumWorkers)
	for i := 0; i < config.NumWorkers; i++ {
		workers[i] = &Worker{
			InputChannel:         make(chan *TaskAndStrategy),
			OutputChannel:        make(chan WorkerResult),
			HandlerInputChannel:  make(chan *TaskAndStrategy),
			HandlerOutputChannel: make(chan WorkerResult),
			TaskTimeout:          config.WorkerTaskTimeout,
		}
		workers[i].Start()
		availableWorkers <- workers[i]
	}

	return &WorkerManager{
		id:                  id,
		config:              config,
		availableWorkers:    availableWorkers,
		workers:             workers,
		inputChannel:        make(chan []*Message),
		currentBatch:        newTaskBatch(),
		batchOrder:          make([]TaskId, 0),
		topicPartition:      topicPartition,
		largestOffset:       InvalidOffset,
		lastCommittedOffset: InvalidOffset,
		failCounter:         NewFailureCounter(config.WorkerRetryThreshold, config.WorkerThresholdTimeWindow),
		batchProcessed:      make(chan bool),
		managerStop:         make(chan bool),
		processingStop:      make(chan bool),
		commitStop:          make(chan bool),
		metrics:             metrics,
		closeConsumer:       closeConsumer,
	}
}
327621
func NewWorker(config Config) (worker.Worker, error) {
	if err := config.Validate(); err != nil {
		return nil, errors.Trace(err)
	}

	rateLimitConfig, err := getRateLimitConfig(config.AgentConfig)
	if err != nil {
		return nil, errors.Annotate(err, "getting rate limit config")
	}

	logSinkConfig, err := getLogSinkConfig(config.AgentConfig)
	if err != nil {
		return nil, errors.Annotate(err, "getting log sink config")
	}

	controllerConfig, err := config.StatePool.SystemState().ControllerConfig()
	if err != nil {
		return nil, errors.Annotate(err, "cannot fetch the controller config")
	}

	observerFactory, err := newObserverFn(
		config.AgentConfig,
		controllerConfig,
		config.Clock,
		config.Hub,
		config.MetricsCollector,
	)
	if err != nil {
		return nil, errors.Annotate(err, "cannot create RPC observer factory")
	}

	serverConfig := apiserver.ServerConfig{
		StatePool:                     config.StatePool,
		Controller:                    config.Controller,
		Clock:                         config.Clock,
		Tag:                           config.AgentConfig.Tag(),
		DataDir:                       config.AgentConfig.DataDir(),
		LogDir:                        config.AgentConfig.LogDir(),
		Hub:                           config.Hub,
		Presence:                      config.Presence,
		Mux:                           config.Mux,
		Authenticator:                 config.Authenticator,
		RestoreStatus:                 config.RestoreStatus,
		UpgradeComplete:               config.UpgradeComplete,
		PublicDNSName:                 controllerConfig.AutocertDNSName(),
		AllowModelAccess:              controllerConfig.AllowModelAccess(),
		NewObserver:                   observerFactory,
		RegisterIntrospectionHandlers: config.RegisterIntrospectionHTTPHandlers,
		MetricsCollector:              config.MetricsCollector,
		RateLimitConfig:               rateLimitConfig,
		LogSinkConfig:                 &logSinkConfig,
		GetAuditConfig:                config.GetAuditConfig,
		LeaseManager:                  config.LeaseManager,
	}
	return config.NewServer(serverConfig)
}
172019
func NewGCWorker(store tikv.Storage, pdClient pd.Client) (tikv.GCHandler, error) {
	ver, err := store.CurrentVersion()
	if err != nil {
		return nil, errors.Trace(err)
	}
	hostName, err := os.Hostname()
	if err != nil {
		hostName = "unknown"
	}
	worker := &GCWorker{
		uuid:        strconv.FormatUint(ver.Ver, 16),
		desc:        fmt.Sprintf("host:%s, pid:%d, start at %s", hostName, os.Getpid(), time.Now()),
		store:       store,
		pdClient:    pdClient,
		gcIsRunning: false,
		lastFinish:  time.Now(),
		done:        make(chan error),
	}
	return worker, nil
}
1255068
func NewRetryStrategyWorker(config WorkerConfig) (worker.Worker, error) {
	if err := config.Validate(); err != nil {
		return nil, errors.Trace(err)
	}
	w, err := watcher.NewNotifyWorker(watcher.NotifyConfig{
		Handler: retryStrategyHandler{config},
	})
	if err != nil {
		return nil, errors.Trace(err)
	}
	return &RetryStrategyWorker{w, config.RetryStrategy}, nil
}
989978
func NewConnectedStatusWorker(cfg ConnectedConfig) (worker.Worker, error) {
	handler, err := NewConnectedStatusHandler(cfg)
	if err != nil {
		return nil, errors.Trace(err)
	}
	return watcher.NewNotifyWorker(watcher.NotifyConfig{
		Handler: handler,
	})
}
459448
func newWorker(errorHandler func(error) error, closing chan bool, done chan bool) *worker {
	w := &worker{
		lock:         &sync.Mutex{},
		sink:         NewHTTPSink(),
		errorHandler: errorHandler,
		closing:      closing,
		done:         done,
	}

	return w
}
895163
func newWaitWorker(config Config, targetVersion int) (worker.Worker, error) {
	watcher, err := config.Facade.WatchModelEnvironVersion(config.ModelTag)
	if err != nil {
		return nil, errors.Trace(err)
	}
	ww := waitWorker{
		watcher:       watcher,
		facade:        config.Facade,
		modelTag:      config.ModelTag,
		gate:          config.GateUnlocker,
		targetVersion: targetVersion,
	}
	if err := catacomb.Invoke(catacomb.Plan{
		Site: &ww.catacomb,
		Init: []worker.Worker{watcher},
		Work: ww.loop,
	}); err != nil {
		return nil, errors.Trace(err)
	}
	return &ww, nil
}
104327
func New(conf pruner.Config) (worker.Worker, error) {
	if err := conf.Validate(); err != nil {
		return nil, errors.Trace(err)
	}

	w := &Worker{
		pruner.New(conf),
	}

	err := catacomb.Invoke(catacomb.Plan{
		Site: w.Catacomb(),
		Work: w.loop,
	})

	return w, errors.Trace(err)
}
145534
func NewWorker(srv *Server) (*Worker, error) {
	w := &Worker{
		srv:    srv,
		logger: srv.logger.ResetNamed("worker"),
		start:  time.Now(),
	}
	w.pauseCond = sync.NewCond(&w.pauseLock)
	go w.run()
	return w, nil
}
477219
func NewBlockWorker(wr *wrangler.Wrangler) (Worker, error) {
	return &BlockWorker{
		StatusWorker: NewStatusWorker(),
		wr:           wr,
	}, nil
}
1142857
func newWorkerPool(tasks []*workTask, concurrency int) *workerPool {
	return &workerPool{
		Tasks:       tasks,
		concurrency: concurrency,
		tasksChan:   make(chan *workTask),
	}
}