// NewWorker returns a new API server worker, with the given configuration.
493923
func NewWorker(config Config) (worker.Worker, error) {
	if err := config.Validate(); err != nil {
		return nil, errors.Trace(err)
	}

	w := &Worker{
		logger: config.Logger,
		update: config.UpdateControllerAPIPort,
	}

	servingInfo, ok := config.AgentConfig.StateServingInfo()
	if !ok {
		return nil, errors.New("missing state serving info")
	}
	w.controllerAPIPort = servingInfo.ControllerAPIPort
	// We need to make sure that update the agent config with
	// the potentially new controller api port from the database
	// before we start any other workers that need to connect
	// to the controller over the controller port.
	w.updateControllerPort(config.ControllerAPIPort)

	unsub, err := config.Hub.Subscribe(controller.ConfigChanged,
		func(topic string, data controller.ConfigChangedMessage, err error) {
			if w.updateControllerPort(data.Config.ControllerAPIPort()) {
				w.tomb.Kill(dependency.ErrBounce)
			}
		})
	if err != nil {
		return nil, errors.Annotate(err, "unable to subscribe to details topic")
	}

	w.tomb.Go(func() error {
		defer unsub()
		<-w.tomb.Dying()
		return nil
	})
	return w, 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
}
1353709
func NewWorker(config Config) (worker.Worker, error) {
	if err := config.Validate(); err != nil {
		return nil, errors.Trace(err)
	}
	if runtime.GOOS != "linux" {
		return nil, errors.NotSupportedf("os %q", runtime.GOOS)
	}

	path := "@" + config.SocketName
	addr, err := net.ResolveUnixAddr("unix", path)
	if err != nil {
		return nil, errors.Annotate(err, "unable to resolve unix socket")
	}

	l, err := net.ListenUnix("unix", addr)
	if err != nil {
		return nil, errors.Annotate(err, "unable to listen on unix socket")
	}
	logger.Debugf("introspection worker listening on %q", path)

	w := &socketListener{
		listener:           l,
		depEngine:          config.DepEngine,
		statePool:          config.StatePool,
		pubsub:             config.PubSub,
		machineLock:        config.MachineLock,
		prometheusGatherer: config.PrometheusGatherer,
		presence:           config.Presence,
		done:               make(chan struct{}),
	}
	go w.serve()
	w.tomb.Go(w.run)
	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,
	}
}
92857
func NewHandler(c Config) func(http.Handler) http.Handler {
	if c.Output == nil {
		c.Output = NewOutputChannel(NewConsoleOutput())
	}
	return func(next http.Handler) http.Handler {
		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			var l Logger
			if r != nil {
				l = New(c)
				r = r.WithContext(NewContext(r.Context(), l))
			}
			next.ServeHTTP(w, r)
			if l, ok := l.(*logger); ok {
				l.close()
			}
		})
	}
}
933222
func (server *Server) NewWorker(consumerTag string, concurrency int) *Worker {
	return &Worker{
		server:      server,
		ConsumerTag: consumerTag,
		Concurrency: concurrency,
		Queue:       "",
	}
}
242679
func NewCleanupWorker(w worker.Worker, cleanup func()) worker.Worker {
	return &CleanupWorker{
		Worker:  w,
		cleanup: cleanup,
	}
}
1259689
func (config ManifoldConfig) newWorker(a agent.Agent, apiCaller base.APICaller) (worker.Worker, error) {
	cfg := a.CurrentConfig()
	// Grab the tag and ensure that it's for a machine.
	tag, ok := cfg.Tag().(names.MachineTag)
	if !ok {
		return nil, errors.New("agent's tag is not a machine tag")
	}

	// Get the machine agent's jobs.
	// TODO(fwereade): this functionality should be on the
	// deployer facade instead.
	agentFacade, err := apiagent.NewState(apiCaller)
	if err != nil {
		return nil, errors.Trace(err)
	}
	entity, err := agentFacade.Entity(tag)
	if err != nil {
		return nil, errors.Trace(err)
	}

	var isUnitHoster bool
	for _, job := range entity.Jobs() {
		if job == multiwatcher.JobHostUnits {
			isUnitHoster = true
			break
		}
	}

	if !isUnitHoster {
		return nil, dependency.ErrUninstall
	}

	deployerFacade := apideployer.NewState(apiCaller)
	context := config.NewDeployContext(deployerFacade, cfg)
	w, err := NewDeployer(deployerFacade, context)
	if err != nil {
		return nil, errors.Annotate(err, "cannot start unit agent deployer worker")
	}
	return w, nil
}
286328
func NewDoer(client *http.Client, userAgent string) *Doer {
	return &Doer{
		httpClient: client,
		userAgent:  userAgent,
	}
}
1069967
func newWorker(a agent.Agent, apiCaller base.APICaller) (worker.Worker, error) {
	currentConfig := a.CurrentConfig()

	// TODO(fwereade): this functionality should be on the
	// machiner facade instead -- or, better yet, separate
	// the networking concerns from the lifecycle ones and
	// have completely separate workers.
	//
	// (With their own facades.)
	agentFacade, err := apiagent.NewState(apiCaller)
	if err != nil {
		return nil, errors.Trace(err)
	}
	modelConfig, err := agentFacade.ModelConfig()
	if err != nil {
		return nil, errors.Errorf("cannot read environment config: %v", err)
	}

	ignoreMachineAddresses, _ := modelConfig.IgnoreMachineAddresses()
	// Containers only have machine addresses, so we can't ignore them.
	tag := currentConfig.Tag()
	if names.IsContainerMachine(tag.Id()) {
		ignoreMachineAddresses = false
	}
	if ignoreMachineAddresses {
		logger.Infof("machine addresses not used, only addresses from provider")
	}
	accessor := APIMachineAccessor{apimachiner.NewState(apiCaller)}
	w, err := NewMachiner(Config{
		MachineAccessor:              accessor,
		Tag:                          tag.(names.MachineTag),
		ClearMachineAddressesOnStart: ignoreMachineAddresses,
		NotifyMachineDead: func() error {
			return agent.SetCanUninstall(a)
		},
	})
	if err != nil {
		return nil, errors.Annotate(err, "cannot start machiner worker")
	}
	return w, err
}
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
}
907863
func newExecWorker(probeCmd, probePath string, execPeriod time.Duration, exec utilexec.Interface, clock clock.Clock, readyCh chan<- struct{}) *execWorker {
	return &execWorker{
		// Initializing the result with a timestamp here allows us to
		// wait maxLatency for the worker goroutine to start, and for each
		// iteration of the worker to complete.
		exec:      exec,
		clock:     clock,
		result:    execResult{[]byte{}, nil, clock.Now()},
		period:    execPeriod,
		probeCmd:  probeCmd,
		probePath: probePath,
		stopCh:    make(chan struct{}),
		readyCh:   readyCh,
	}
}
850251
func NewEnvironWorker(config Config) (worker.Worker, error) {
	config.GetMachineWatcher = config.Facade.WatchMachines
	config.GetRequiredLXDProfiles = func(modelName string) []string {
		return []string{"default", "juju-" + modelName}
	}
	return newWorker(config)
}
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
}
1170148
func NewConfigurator(config Config, logger *log.Logger) (*Configurator, error) {
	c := &Configurator{logger: logger}
	err := c.Update(config)
	if err != nil {
		return nil, err
	}
	return c, nil
}
1331234
func NewMinUnitsWorker(st *state.State) worker.Worker {
	mu := &MinUnitsWorker{st: st}
	return legacy.NewStringsWorker(mu)
}
1296680
func NewIsolatedStatusWorker(cfg IsolatedConfig) (worker.Worker, error) {
	if err := cfg.Validate(); err != nil {
		return nil, errors.Trace(err)
	}
	w := &isolatedStatusWorker{
		config: cfg,
	}
	w.tomb.Go(w.loop)
	return w, nil
}