// NewWorker returns a new API server worker, with the given configuration.
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)
}
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
}
508788
func NewWorker(api Facade, env environs.Environ, credentialAPI common.CredentialAPI) (worker.Worker, error) {
	envNetworking, _ := environs.SupportsNetworking(env)
	w, err := watcher.NewNotifyWorker(watcher.NotifyConfig{
		Handler: &Undertaker{
			API:         api,
			Releaser:    envNetworking,
			CallContext: common.NewCloudCallContext(credentialAPI, nil),
		},
	})
	if err != nil {
		return nil, errors.Trace(err)
	}
	return w, nil
}
1092071
func NewWorker(config *config.Config) (*Worker, error) {
	hostname, err := os.Hostname()
	if err != nil {
		return nil, err
	}

	workers.Configure(map[string]string{
		"server":   config.GetString("pitaya.worker.redis.url"),
		"pool":     config.GetString("pitaya.worker.redis.pool"),
		"password": config.GetString("pitaya.worker.redis.password"),
		"process":  hostname,
	})

	return &Worker{
		concurrency: config.GetInt("pitaya.worker.concurrency"),
		opts:        NewEnqueueOpts(config),
		config:      config,
	}, nil
}
1062932
func NewWorker() (w *Worker) {
	astilog.Info("astiworker: starting worker...")
	w = &Worker{wg: &sync.WaitGroup{}}
	w.ctx, w.cancel = context.WithCancel(context.Background())
	w.wg.Add(1)
	return
}
455303
func NewHandler(Next httpserver.Handler, config *Config) *Handler {
	return &Handler{
		Config:   config,
		Cache:    NewHTTPCache(config.CacheKeyTemplate),
		URLLocks: NewURLLock(),
		Next:     Next,
	}
}
170005
func New(config Config) (*Worker, error) {
	if err := config.Validate(); err != nil {
		return nil, errors.Trace(err)
	}
	phase, err := config.Facade.Phase(config.Model)
	if err != nil {
		return nil, errors.Trace(err)
	}

	w := &Worker{
		config: config,
		phase:  phase,
	}
	err = catacomb.Invoke(catacomb.Plan{
		Site: &w.catacomb,
		Work: w.loop,
	})
	if err != nil {
		return nil, errors.Trace(err)
	}
	return w, nil
}
545134
func NewCeleryWorker(broker CeleryBroker, backend CeleryBackend, numWorkers int) *CeleryWorker {
	return &CeleryWorker{
		broker:          broker,
		backend:         backend,
		numWorkers:      numWorkers,
		registeredTasks: map[string]interface{}{},
		rateLimitPeriod: 100 * time.Millisecond,
	}
}
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
}
1053086
func 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("this manifold may only be used inside a machine agent")
	}

	// Get the machine agent's jobs.
	apiSt, err := apiagent.NewState(apiCaller)
	if err != nil {
		return nil, errors.Trace(err)
	}
	entity, err := apiSt.Entity(tag)
	if err != nil {
		return nil, err
	}
	var isModelManager bool
	for _, job := range entity.Jobs() {
		if job == multiwatcher.JobManageModel {
			isModelManager = true
			break
		}
	}

	if !isModelManager {
		return nil, dependency.ErrMissing
	}

	return NewWorker(cfg)
}
591253
func NewWorker(config Config) (worker.Worker, error) {
	value, err := config.Value()
	if err != nil {
		return nil, errors.Trace(err)
	}
	flag := &Worker{
		config: config,
		value:  value,
	}
	if err := catacomb.Invoke(catacomb.Plan{
		Site: &flag.catacomb,
		Work: flag.loop,
	}); err != nil {
		return nil, errors.Trace(err)
	}
	return flag, nil
}
933222
func (server *Server) NewWorker(consumerTag string, concurrency int) *Worker {
	return &Worker{
		server:      server,
		ConsumerTag: consumerTag,
		Concurrency: concurrency,
		Queue:       "",
	}
}
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
}
1250318
func NewLogger(api LoggerAPI, tag names.Tag, loggingOverride string, updateCallback func(string) error) (worker.Worker, error) {
	logger := &Logger{
		api:            api,
		tag:            tag,
		updateCallback: updateCallback,
		lastConfig:     loggo.LoggerInfo(),
		configOverride: loggingOverride,
	}
	log.Debugf("initial log config: %q", logger.lastConfig)

	w, err := watcher.NewNotifyWorker(watcher.NotifyConfig{
		Handler: logger,
	})
	if err != nil {
		return nil, errors.Trace(err)
	}
	return w, nil
}
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
}
836784
func newBulkWorker(p *BulkProcessor, i int) *bulkWorker {
	return &bulkWorker{
		p:           p,
		i:           i,
		bulkActions: p.bulkActions,
		bulkSize:    p.bulkSize,
		service:     NewBulkService(p.c),
		flushC:      make(chan struct{}),
		flushAckC:   make(chan struct{}),
	}
}
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)
}