// initializeWatchers is responsible for creating the watchers
498776
func initializeWatchers(client kubernetes.Interface, dnsctl *dns.DNSController, namespace string, watchIngress bool) error {
	glog.V(1).Infof("initializing the watch controllers, namespace: %q", namespace)

	nodeController, err := watchers.NewNodeController(client, dnsctl)
	if err != nil {
		return fmt.Errorf("failed to initialize the node controller, error: %v", err)
	}

	podController, err := watchers.NewPodController(client, dnsctl, namespace)
	if err != nil {
		return fmt.Errorf("failed to initialize the pod controller, error: %v", err)
	}

	serviceController, err := watchers.NewServiceController(client, dnsctl, namespace)
	if err != nil {
		return fmt.Errorf("failed to initialize the service controller, error: %v", err)
	}

	var ingressController *watchers.IngressController
	if watchIngress {
		ingressController, err = watchers.NewIngressController(client, dnsctl, namespace)
		if err != nil {
			return fmt.Errorf("failed to initialize the ingress controller, error: %v", err)
		}
	} else {
		glog.Infof("Ingress controller disabled")
	}

	go nodeController.Run()
	go podController.Run()
	go serviceController.Run()

	if watchIngress {
		go ingressController.Run()
	}

	return nil
}
1093158
func (c *consulClient) ListAndWatch(name, prefix string, chanSize int) *Watcher {
	w := newWatcher(name, prefix, chanSize)

	log.WithField(fieldWatcher, w).Debug("Starting watcher...")

	go c.Watch(w)

	return w
}
1312652
func (s *Sanitizer) NewWatcher(ctx context.Context, watch Watch) (Watcher, error) {
	for _, prefix := range watch.Prefixes {
		if !isKeySafe(prefix) {
			return nil, trace.BadParameter(errorMessage)
		}
	}
	return s.backend.NewWatcher(ctx, watch)
}
1200128
func newWatcher(c chan<- EventInfo) watcher {
	i := &inotify{
		m:      make(map[int32]*watched),
		fd:     invalidDescriptor,
		pipefd: []int{invalidDescriptor, invalidDescriptor},
		epfd:   invalidDescriptor,
		epes:   make([]unix.EpollEvent, 0),
		c:      c,
	}
	runtime.SetFinalizer(i, func(i *inotify) {
		i.epollclose()
		if i.fd != invalidDescriptor {
			unix.Close(int(i.fd))
		}
	})
	return i
}
339584
func (m *Machine) WatchContainers() (watcher.StringsWatcher, error) {
	var result params.StringsWatchResult
	arg := params.Entity{Tag: m.tag.String()}
	err := m.facade.FacadeCall("WatchContainers", arg, &result)
	if err != nil {
		return nil, errors.Trace(err)
	}
	if result.Error != nil {
		return nil, result.Error
	}
	return apiwatcher.NewStringsWatcher(m.facade.RawAPICaller(), result), nil
}
1338396
func (api *InstanceMutaterAPI) WatchMachines() (params.StringsWatchResult, error) {
	result := params.StringsWatchResult{}
	if !api.authorizer.AuthController() {
		return result, common.ErrPerm
	}

	watch, err := api.model.WatchMachines()
	if err != nil {
		return result, err
	}
	if changes, ok := <-watch.Changes(); ok {
		result.StringsWatcherId = api.resources.Register(watch)
		result.Changes = changes
	} else {
		return result, errors.Errorf("cannot obtain initial model machines")
	}
	return result, nil
}
238718
func (k *kubernetesClient) WatchService(appName string) (watcher.NotifyWatcher, error) {
	// Application may be a statefulset or deployment. It may not have
	// been set up when the watcher is started so we don't know which it
	// is ahead of time. So use a multi-watcher to cover both cases.
	statefulsets := k.client().AppsV1().StatefulSets(k.namespace)
	sswatcher, err := statefulsets.Watch(v1.ListOptions{
		LabelSelector: applicationSelector(appName),
		Watch:         true,
	})
	if err != nil {
		return nil, errors.Trace(err)
	}
	w1, err := k.newWatcher(sswatcher, appName, k.clock)
	if err != nil {
		return nil, errors.Trace(err)
	}

	deployments := k.client().AppsV1().Deployments(k.namespace)
	dwatcher, err := deployments.Watch(v1.ListOptions{
		LabelSelector: applicationSelector(appName),
		Watch:         true,
	})
	if err != nil {
		return nil, errors.Trace(err)
	}
	w2, err := k.newWatcher(dwatcher, appName, k.clock)
	if err != nil {
		return nil, errors.Trace(err)
	}

	return watcher.NewMultiNotifyWatcher(w1, w2), nil
}
981285
func (eventBroadcaster *eventBroadcasterImpl) StartEventWatcher(eventHandler func(*v1.Event)) watch.Interface {
	watcher := eventBroadcaster.Watch()
	go func() {
		defer utilruntime.HandleCrash()
		for watchEvent := range watcher.ResultChan() {
			event, ok := watchEvent.Object.(*v1.Event)
			if !ok {
				// This is all local, so there's no reason this should
				// ever happen.
				continue
			}
			eventHandler(event)
		}
	}()
	return watcher
}
178202
func (db *BytesConnectionRedis) NewWatcher(prefix string) keyval.BytesWatcher {
	return db.NewBrokerWatcher(prefix)
}
796831
func NewWatcher(sessionStartTime time.Time) *Watcher {
	return &Watcher{
		sessionStartTime: sessionStartTime,
		errorChan:        make(chan *probe.Error),
		eventInfoChan:    make(chan EventInfo),
		o:                []*watchObject{},
	}
}
902067
func NewWatcher() (*Watcher, error) {
	port, e := syscall.CreateIoCompletionPort(syscall.InvalidHandle, 0, 0, 0)
	if e != nil {
		return nil, os.NewSyscallError("CreateIoCompletionPort", e)
	}
	w := &Watcher{
		port:    port,
		watches: make(watchMap),
		input:   make(chan *input, 1),
		Events:  make(chan Event, 50),
		Errors:  make(chan error),
		quit:    make(chan chan<- error, 1),
	}
	go w.readEvents()
	return w, nil
}
503337
func (c *CircularBuffer) NewWatcher(ctx context.Context, watch Watch) (Watcher, error) {
	c.Lock()
	defer c.Unlock()

	select {
	case <-c.ctx.Done():
		return nil, trace.BadParameter("buffer is closed")
	default:
	}

	if watch.QueueSize == 0 {
		watch.QueueSize = len(c.events)
	}

	if len(watch.Prefixes) == 0 {
		// if watcher has no prefixes, assume it will match anything
		// starting from the separator (what includes all keys in backend invariant, see Keys function)
		watch.Prefixes = append(watch.Prefixes, []byte{Separator})
	} else {
		// if watcher's prefixes are redundant, keep only shorter prefixes
		// to avoid double fan out
		watch.Prefixes = removeRedundantPrefixes(watch.Prefixes)
	}

	closeCtx, cancel := context.WithCancel(ctx)
	w := &BufferWatcher{
		Watch:    watch,
		eventsC:  make(chan Event, watch.QueueSize),
		ctx:      closeCtx,
		cancel:   cancel,
		capacity: watch.QueueSize,
	}
	c.Debugf("Add %v.", w)
	select {
	case w.eventsC <- Event{Type: OpInit}:
	case <-c.ctx.Done():
		return nil, trace.BadParameter("buffer is closed")
	default:
		c.Warningf("Closing %v, buffer overflow.", w)
		w.Close()
		return nil, trace.BadParameter("buffer overflow")
	}
	c.watchers.add(w)
	return w, nil
}
1264528
func NewWatcher() (*Watcher, error) {
	fd, errno := syscall.InotifyInit()
	if fd == -1 {
		return nil, os.NewSyscallError("inotify_init", errno)
	}
	w := &Watcher{
		fd:            fd,
		watches:       make(map[string]*watch),
		fsnFlags:      make(map[string]uint32),
		paths:         make(map[int]string),
		internalEvent: make(chan *FileEvent),
		Event:         make(chan *FileEvent),
		Error:         make(chan error),
		done:          make(chan bool, 1),
	}

	go w.readEvents()
	go w.purgeEvents()
	return w, nil
}
1117251
func (e *EventsService) NewWatcher(ctx context.Context, watch services.Watch) (services.Watcher, error) {
	if len(watch.Kinds) == 0 {
		return nil, trace.BadParameter("global watches are not supported yet")
	}
	var parsers []resourceParser
	var prefixes [][]byte
	for _, kind := range watch.Kinds {
		if kind.Name != "" && kind.Kind != services.KindNamespace {
			return nil, trace.BadParameter("watch with Name is only supported for Namespace resource")
		}
		var parser resourceParser
		switch kind.Kind {
		case services.KindCertAuthority:
			parser = newCertAuthorityParser(kind.LoadSecrets)
		case services.KindToken:
			parser = newProvisionTokenParser()
		case services.KindStaticTokens:
			parser = newStaticTokensParser()
		case services.KindClusterConfig:
			parser = newClusterConfigParser()
		case services.KindClusterName:
			parser = newClusterNameParser()
		case services.KindNamespace:
			parser = newNamespaceParser(kind.Name)
		case services.KindRole:
			parser = newRoleParser()
		case services.KindUser:
			parser = newUserParser()
		case services.KindNode:
			parser = newNodeParser()
		case services.KindProxy:
			parser = newProxyParser()
		case services.KindAuthServer:
			parser = newAuthServerParser()
		case services.KindTunnelConnection:
			parser = newTunnelConnectionParser()
		case services.KindReverseTunnel:
			parser = newReverseTunnelParser()
		default:
			return nil, trace.BadParameter("watcher on object kind %v is not supported", kind)
		}
		prefixes = append(prefixes, parser.prefix())
		parsers = append(parsers, parser)
	}
	// sort so that longer prefixes get first
	sort.Slice(parsers, func(i, j int) bool { return len(parsers[i].prefix()) > len(parsers[j].prefix()) })
	w, err := e.backend.NewWatcher(ctx, backend.Watch{
		Name:            watch.Name,
		Prefixes:        prefixes,
		QueueSize:       watch.QueueSize,
		MetricComponent: watch.MetricComponent,
	})
	if err != nil {
		return nil, trace.Wrap(err)
	}
	return newWatcher(w, e.Entry, parsers), nil
}
470564
func NewNotifyWatcher(caller base.APICaller, result params.NotifyWatchResult) watcher.NotifyWatcher {
	w := &notifyWatcher{
		caller:          caller,
		notifyWatcherId: result.NotifyWatcherId,
		out:             make(chan struct{}),
	}
	w.tomb.Go(w.loop)
	return w
}
1066929
func (s *Wrapper) NewWatcher(ctx context.Context, watch Watch) (Watcher, error) {
	if err := s.GetReadError(); err != nil {
		return nil, trace.Wrap(err)
	}
	return s.backend.NewWatcher(ctx, watch)
}
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)
}