// WatchAllContainers starts a StringsWatcher to watch all containers deployed to
// any machine passed in args.
323637
func (p *ProvisionerAPI) WatchAllContainers(args params.WatchContainers) (params.StringsWatchResults, error) {
	return p.WatchContainers(args)
}
1020216
func (m *Machine) WatchContainers(ctype instance.ContainerType) StringsWatcher {
	isChild := fmt.Sprintf("^%s/%s/%s$", m.doc.DocID, ctype, names.NumberSnippet)
	return m.containersWatcher(isChild)
}
1275909
func (a *Application) WatchUnits() StringsWatcher {
	members := bson.D{{"application", a.doc.Name}}
	prefix := a.doc.Name + "/"
	filter := func(unitDocID interface{}) bool {
		unitName, err := a.st.strictLocalID(unitDocID.(string))
		if err != nil {
			return false
		}
		return strings.HasPrefix(unitName, prefix)
	}
	return newLifecycleWatcher(a.st, unitsC, members, filter, nil)
}
458890
func (s *SvcProvider) WatchAll(rsps chan core.WatchState) error {
	return s.StateDriver.WatchAllState(svcProviderPathPrefix, s, json.Unmarshal,
		rsps)
}
1196610
func (d *ConsulStateDriver) WatchAll(baseKey string, rsps chan [2][]byte) error {
	baseKey = processKey(baseKey)
	consulRsps := make(chan api.KVPairs, 1)
	stop := make(chan bool, 1)
	recvErr := make(chan error, 2)

	// Consul returns all the keys as return value of List(). The following maps helps
	// track the state that has been seen and used to appropriately generate
	// create, modify and delete events
	kvCache := map[string]*api.KVPair{}
	// read with index=0 to fetch all existing keys
	var waitIndex uint64
	kvs, qm, err := d.Client.KV().List(baseKey, &api.QueryOptions{WaitIndex: waitIndex})
	if err != nil {
		log.Errorf("consul read failed for key %q. Error: %s", baseKey, err)
		return err
	}
	// Consul returns success and a nil kv when a key is not found.
	// Treat this as starting with no state.
	// XXX: shall we fail the watch in this case?
	if kvs == nil {
		kvs = api.KVPairs{}
	}
	for _, kv := range kvs {
		kvCache[kv.Key] = kv
	}
	waitIndex = qm.LastIndex

	go d.channelConsulEvents(baseKey, kvCache, consulRsps, rsps, recvErr, stop)

	for {
		select {
		case err := <-recvErr:
			return err
		default:
			kvs, qm, err := d.Client.KV().List(baseKey, &api.QueryOptions{WaitIndex: waitIndex})
			if err != nil {
				if api.IsServerError(err) || strings.Contains(err.Error(), "EOF") || strings.Contains(err.Error(), "connection refused") {
					log.Warnf("Consul watch: server error: %v for %s. Retrying..", err, baseKey)
					time.Sleep(5 * time.Second)
					continue
				} else {
					log.Errorf("consul watch failed for key %q. Error: %s. stopping watch..", baseKey, err)
					stop <- true
					return err
				}
			}
			// Consul returns success and a nil kv when a key is not found.
			// This shall translate into appropriate 'Delete' events or
			// no events (depending on whether some keys were seen before)
			// XXX: shall we stop the watch in this case?
			if kvs == nil {
				kvs = api.KVPairs{}
			}

			waitIndex = qm.LastIndex
			consulRsps <- kvs
		}
	}
}
1200579
func (st *State) WatchApplications() StringsWatcher {
	return newLifecycleWatcher(st, applicationsC, nil, isLocalID(st), nil)
}
54635
func (s *EndpointGroupState) WatchAll(rsps chan core.WatchState) error {
	return s.StateDriver.WatchAllState(epGroupConfigPathPrefix, s, json.Unmarshal,
		rsps)
}
228787
func (sb *storageBackend) WatchMachineFilesystems(m names.MachineTag) StringsWatcher {
	return sb.watchHostStorage(m, filesystemsC)
}
96114
func (api *API) Watch() (watcher.StringsWatcher, error) {
	var result params.StringsWatchResult
	err := api.caller.FacadeCall("Watch", nil, &result)
	if err != nil {
		return nil, errors.Trace(err)
	}
	if result.Error != nil {
		return nil, errors.Trace(result.Error)
	}
	w := api.newWatcher(api.caller.RawAPICaller(), result)
	return w, nil
}
274539
func (s *StorageProvisionerAPIv3) WatchFilesystemAttachments(args params.Entities) (params.MachineStorageIdsWatchResults, error) {
	w := filesystemwatcher.Watchers{s.sb}
	return s.watchAttachments(
		args,
		w.WatchModelManagedFilesystemAttachments,
		w.WatchMachineManagedFilesystemAttachments,
		w.WatchUnitManagedFilesystemAttachments,
		storagecommon.ParseFilesystemAttachmentIds,
	)
}
763708
func servicesWatch(params map[string]interface{}) (WatcherFunc, error) {
	stale := false
	if err := assignValueBool(params, "stale", &stale); err != nil {
		return nil, err
	}

	fn := func(p *Plan) (BlockingParamVal, interface{}, error) {
		catalog := p.client.Catalog()
		opts := makeQueryOptionsWithContext(p, stale)
		defer p.cancelFunc()
		services, meta, err := catalog.Services(&opts)
		if err != nil {
			return nil, nil, err
		}
		return WaitIndexVal(meta.LastIndex), services, err
	}
	return fn, nil
}
1183014
func (st *State) WatchVolumes(scope names.Tag) (watcher.StringsWatcher, error) {
	return st.watchStorageEntities("WatchVolumes", scope)
}
179866
func (st *State) WatchModels() StringsWatcher {
	return newCollectionWatcher(st, colWCfg{
		col:    modelsC,
		global: true,
	})
}
1059790
func (g *GRPCServer) WatchEvents(watch *proto.Watch, stream proto.AuthService_WatchEventsServer) error {
	auth, err := g.authenticate(stream.Context())
	if err != nil {
		return trail.ToGRPC(err)
	}
	servicesWatch := services.Watch{
		Name: auth.User.GetName(),
	}
	for _, kind := range watch.Kinds {
		servicesWatch.Kinds = append(servicesWatch.Kinds, services.WatchKind{
			Name:        kind.Name,
			Kind:        kind.Kind,
			LoadSecrets: kind.LoadSecrets,
		})
	}
	watcher, err := auth.NewWatcher(stream.Context(), servicesWatch)
	if err != nil {
		return trail.ToGRPC(err)
	}
	defer watcher.Close()

	for {
		select {
		case <-stream.Context().Done():
			return nil
		case <-watcher.Done():
			return trail.ToGRPC(watcher.Error())
		case event := <-watcher.Events():
			out, err := eventToGRPC(event)
			if err != nil {
				return trail.ToGRPC(err)
			}
			if err := stream.Send(out); err != nil {
				return trail.ToGRPC(err)
			}
		}
	}
}
73192
func (r *Containerd) ListContainers(filter string) ([]string, error) {
	return listCRIContainers(r.Runner, filter)
}
184395
func (r *CRIO) ListContainers(filter string) ([]string, error) {
	return listCRIContainers(r.Runner, filter)
}
1366359
func (st *State) WatchSubnets(subnetFilter func(id interface{}) bool) StringsWatcher {
	filter := func(id interface{}) bool {
		subnet, err := st.strictLocalID(id.(string))
		if err != nil {
			return false
		}
		if subnetFilter == nil {
			return true
		}
		return subnetFilter(subnet)
	}

	return newLifecycleWatcher(st, subnetsC, nil, filter, nil)
}