// NewNodeDrainWatcher returns a new node drain watcher.
321903
func NewNodeDrainWatcher(ctx context.Context, limiter *rate.Limiter, state *state.StateStore, logger log.Logger, tracker NodeTracker) *nodeDrainWatcher {
	w := &nodeDrainWatcher{
		ctx:     ctx,
		limiter: limiter,
		logger:  logger.Named("node_watcher"),
		tracker: tracker,
		state:   state,
	}

	go w.watch()
	return w
}
1078523
func (k *kubernetesClient) WatchOperator(appName string) (watcher.NotifyWatcher, error) {
	pods := k.client().CoreV1().Pods(k.namespace)
	w, err := pods.Watch(v1.ListOptions{
		LabelSelector: operatorSelector(appName),
		Watch:         true,
	})
	if err != nil {
		return nil, errors.Trace(err)
	}
	return k.newWatcher(w, appName, k.clock)
}
35709
func (p *Plugin) NewWatcher(name string) messaging.ProtoWatcher {
	return p.NewProtoConnection(name)
}
136949
func (m *Memory) NewWatcher(ctx context.Context, watch backend.Watch) (backend.Watcher, error) {
	if m.EventsOff {
		return nil, trace.BadParameter("events are turned off for this backend")
	}
	return m.buf.NewWatcher(ctx, watch)
}
956343
func NewCloudWatchReactor(reactor CloudWatchReactor,
	subscriptions map[string]sparta.CloudWatchEventsRule,
	additionalLambdaPermissions []sparta.IAMRolePrivilege) (*sparta.LambdaAWSInfo, error) {
	if len(subscriptions) <= 0 {
		return nil, errors.Errorf("CloudWatchLogs subscription map must not be empty")
	}

	reactorLambda := func(ctx context.Context, cwLogs awsLambdaEvents.CloudwatchLogsEvent) (interface{}, error) {
		return reactor.OnCloudWatchMessage(ctx, cwLogs)
	}
	lambdaFn, lambdaFnErr := sparta.NewAWSLambda(reactorName(reactor),
		reactorLambda,
		sparta.IAMRoleDefinition{})
	if lambdaFnErr != nil {
		return nil, errors.Wrapf(lambdaFnErr, "attempting to create reactor")
	}
	cloudWatchEventsPermission := sparta.CloudWatchEventsPermission{}
	cloudWatchEventsPermission.Rules = make(map[string]sparta.CloudWatchEventsRule)
	for eachRuleName, eachRule := range subscriptions {
		cloudWatchEventsPermission.Rules[eachRuleName] = eachRule
	}
	lambdaFn.Permissions = append(lambdaFn.Permissions, cloudWatchEventsPermission)

	if len(additionalLambdaPermissions) != 0 {
		lambdaFn.RoleDefinition.Privileges = additionalLambdaPermissions
	}
	return lambdaFn, nil
}
35158
func NewWatcher() (*Watcher, error) {
	// Create inotify fd
	fd, errno := unix.InotifyInit1(unix.IN_CLOEXEC)
	if fd == -1 {
		return nil, errno
	}
	// Create epoll
	poller, err := newFdPoller(fd)
	if err != nil {
		unix.Close(fd)
		return nil, err
	}
	w := &Watcher{
		fd:       fd,
		poller:   poller,
		watches:  make(map[string]*watch),
		paths:    make(map[int]string),
		Events:   make(chan Event),
		Errors:   make(chan error),
		done:     make(chan struct{}),
		doneResp: make(chan struct{}),
	}

	go w.readEvents()
	return w, nil
}
869078
func (s *StateStore) BatchUpdateNodeDrain(index uint64, updates map[string]*structs.DrainUpdate, events map[string]*structs.NodeEvent) error {
	txn := s.db.Txn(true)
	defer txn.Abort()
	for node, update := range updates {
		if err := s.updateNodeDrainImpl(txn, index, node, update.DrainStrategy, update.MarkEligible, events[node]); err != nil {
			return err
		}
	}
	txn.Commit()
	return nil
}
737011
func NewCacherListerWatcher(storage storage.Interface, resourcePrefix string, newListFunc func() runtime.Object) cache.ListerWatcher {
	return &cacherListerWatcher{
		storage:        storage,
		resourcePrefix: resourcePrefix,
		newListFunc:    newListFunc,
	}
}
549907
func NewWatcher(sockDir string, deprecatedSockDir string) *Watcher {
	return &Watcher{
		path:           sockDir,
		deprecatedPath: deprecatedSockDir,
		fs:             &utilfs.DefaultFs{},

		handlers:    make(map[string]PluginHandler),
		plugins:     make(map[string]pathInfo),
		pluginsPool: make(map[string]map[string]*sync.Mutex),
	}
}
84983
func NewNodeInformer(client kubernetes.Interface, resyncPeriod time.Duration, indexers cache.Indexers) cache.SharedIndexInformer {
	return NewFilteredNodeInformer(client, resyncPeriod, indexers, nil)
}
688053
func NewEventWatcher() (FileWatcher, error) {
	watcher, err := fsnotify.NewWatcher()
	if err != nil {
		return nil, err
	}
	return &fsNotifyWatcher{watcher}, nil
}
988322
func NewDeadWatcher(err error) *Watcher {
	var w Watcher
	w.tomb.Kill(errors.Trace(err))
	return &w
}
1368164
func newDeploymentWatcher(parent context.Context, queryLimiter *rate.Limiter,
	logger log.Logger, state *state.StateStore, d *structs.Deployment,
	j *structs.Job, triggers deploymentTriggers) *deploymentWatcher {

	ctx, exitFn := context.WithCancel(parent)
	w := &deploymentWatcher{
		queryLimiter:       queryLimiter,
		deploymentID:       d.ID,
		deploymentUpdateCh: make(chan struct{}, 1),
		d:                  d,
		j:                  j,
		state:              state,
		deploymentTriggers: triggers,
		logger:             logger.With("deployment_id", d.ID, "job", j.NamespacedID()),
		ctx:                ctx,
		exitFn:             exitFn,
	}

	// Start the long lived watcher that scans for allocation updates
	go w.watch()

	return w
}
1151301
func (model *Model) WatchForModelConfigChanges() NotifyWatcher {
	return newEntityWatcher(model.st, settingsC, model.st.docID(modelGlobalKey))
}
81971
func (a *AuthServer) NewWatcher(ctx context.Context, watch services.Watch) (services.Watcher, error) {
	return a.GetCache().NewWatcher(ctx, watch)
}
1062813
func GetNodeWatcherFactory() DrainingNodeWatcherFactory {
	return func(ctx context.Context, limiter *rate.Limiter, state *state.StateStore, logger log.Logger, tracker NodeTracker) DrainingNodeWatcher {
		return NewNodeDrainWatcher(ctx, limiter, state, logger, tracker)
	}
}
985948
func NewWatcher(paths *model.RevelContainer, eagerRefresh bool) *Watcher {
	return &Watcher{
		forceRefresh:   true,
		lastError:      -1,
		paths:          paths,
		refreshTimerMS: time.Duration(paths.Config.IntDefault("watch.rebuild.delay", 10)),
		eagerRefresh: eagerRefresh ||
			paths.DevMode &&
				paths.Config.BoolDefault("watch", true) &&
				paths.Config.StringDefault("watch.mode", "normal") == "eager",
		timerMutex:          &sync.Mutex{},
		refreshChannel:      make(chan *utils.Error, 10),
		refreshChannelCount: 0,
	}
}