// WatchContainers starts a StringsWatcher to watch containers deployed to
// any machine passed in args.
14529
func (p *ProvisionerAPI) WatchContainers(args params.WatchContainers) (params.StringsWatchResults, error) {
	result := params.StringsWatchResults{
		Results: make([]params.StringsWatchResult, len(args.Params)),
	}
	for i, arg := range args.Params {
		watcherResult, err := p.watchOneMachineContainers(arg)
		result.Results[i] = watcherResult
		result.Results[i].Error = common.ServerError(err)
	}
	return result, nil
}
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
}
481239
func (r *registry) WatchSubnets(since uint64, stop chan bool) (*subnet.Response, error) {
	for {
		select {
		case event, ok := <-r.events:
			if !ok {
				return nil, errors.New("unexpected close of discoverd event stream")
			}
			if event.Kind != discoverd.EventKindServiceMeta {
				continue
			}
			net := &Network{}
			if err := json.Unmarshal(event.ServiceMeta.Data, net); err != nil {
				return nil, err
			}
			subnets := make(map[string][]byte)
			for subnet, data := range net.Subnets {
				if known, ok := knownSubnets[subnet]; ok && reflect.DeepEqual(known, data) {
					continue
				}
				subnets[subnet] = []byte(*data)
			}
			knownSubnets = net.Subnets
			if event.ServiceMeta.Index >= since {
				return &subnet.Response{Subnets: subnets, Index: event.ServiceMeta.Index}, nil
			}
		case <-stop:
			return nil, nil
		}
	}
}
686950
func (sb *storageBackend) WatchMachineVolumeAttachments(m names.MachineTag) StringsWatcher {
	return sb.watchHostStorageAttachments(m, volumeAttachmentsC)
}
575446
func (t *Topic) WatchPartitions() (PartitionList, <-chan zk.Event, error) {
	value, _, c, err := t.kz.conn.GetW(t.metadataPath())
	if err != nil {
		return nil, nil, err
	}

	list, err := t.parsePartitions(value)
	return list, c, err
}
1220207
func (m *Model) WatchMachines() (*PredicateStringsWatcher, error) {
	defer m.doLocked()()

	// Create a compiled regexp to match machines not containers.
	compiled, err := m.machineRegexp()
	if err != nil {
		return nil, err
	}
	fn := regexpPredicate(compiled)

	// Gather initial slice of machines in this model.
	machines := make([]string, 0)
	for k := range m.machines {
		if fn(k) {
			machines = append(machines, k)
		}
	}

	w := newPredicateStringsWatcher(fn, machines...)
	deregister := m.registerWorker(w)
	unsub := m.hub.Subscribe(m.topic(modelAddRemoveMachine), w.changed)

	w.tomb.Go(func() error {
		<-w.tomb.Dying()
		unsub()
		deregister()
		return nil
	})

	return w, nil
}
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
}
1183014
func (st *State) WatchVolumes(scope names.Tag) (watcher.StringsWatcher, error) {
	return st.watchStorageEntities("WatchVolumes", scope)
}
564225
func (r *Relation) WatchUnits(appName string) (RelationUnitsWatcher, error) {
	return r.watchUnits(appName, false)
}
710944
func (c *Client) WatchUnits(application string) (watcher.StringsWatcher, error) {
	applicationTag, err := applicationTag(application)
	if err != nil {
		return nil, errors.Trace(err)
	}
	args := entities(applicationTag)

	var results params.StringsWatchResults
	if err := c.facade.FacadeCall("WatchUnits", args, &results); err != nil {
		return nil, err
	}
	if n := len(results.Results); n != 1 {
		return nil, errors.Errorf("expected 1 result, got %d", n)
	}
	if err := results.Results[0].Error; err != nil {
		return nil, errors.Trace(err)
	}
	w := apiwatcher.NewStringsWatcher(c.facade.RawAPICaller(), results.Results[0])
	return w, nil
}
202162
func (a *Application) WatchRelations() StringsWatcher {
	return watchApplicationRelations(a.st, a.doc.Name)
}
1317150
func (fw Watchers) WatchModelManagedFilesystems() state.StringsWatcher {
	return newFilteredStringsWatcher(fw.Backend.WatchModelFilesystems(), func(id string) (bool, error) {
		f, err := fw.Backend.Filesystem(names.NewFilesystemTag(id))
		if errors.IsNotFound(err) {
			return false, nil
		} else if err != nil {
			return false, errors.Trace(err)
		}
		_, err = f.Volume()
		return err == state.ErrNoBackingVolume, nil
	})
}
277079
func (loader *TemplateLoader) WatchDir(info os.FileInfo) bool {
	// Watch all directories, except the ones starting with a dot.
	return !strings.HasPrefix(info.Name(), ".")
}
179866
func (st *State) WatchModels() StringsWatcher {
	return newCollectionWatcher(st, colWCfg{
		col:    modelsC,
		global: true,
	})
}
979441
func (u *UnitsWatcher) WatchUnits(args params.Entities) (params.StringsWatchResults, error) {
	result := params.StringsWatchResults{
		Results: make([]params.StringsWatchResult, len(args.Entities)),
	}
	if len(args.Entities) == 0 {
		return result, nil
	}
	canWatch, err := u.getCanWatch()
	if err != nil {
		return params.StringsWatchResults{}, errors.Trace(err)
	}
	for i, entity := range args.Entities {
		tag, err := names.ParseTag(entity.Tag)
		if err != nil {
			result.Results[i].Error = ServerError(ErrPerm)
			continue
		}
		entityResult, err := u.watchOneEntityUnits(canWatch, tag)
		result.Results[i] = entityResult
		result.Results[i].Error = ServerError(err)
	}
	return result, nil
}
1071859
func (c *Client) WatchActionNotifications(agent names.MachineTag) (watcher.StringsWatcher, error) {
	var results params.StringsWatchResults
	args := params.Entities{
		Entities: []params.Entity{{Tag: agent.String()}},
	}

	err := c.facade.FacadeCall("WatchActionNotifications", args, &results)
	if err != nil {
		return nil, errors.Trace(err)
	}

	if len(results.Results) != 1 {
		return nil, errors.Errorf("expected 1 result, got %d", len(results.Results))
	}

	result := results.Results[0]
	if result.Error != nil {
		return nil, errors.Trace(result.Error)
	}
	w := apiwatcher.NewStringsWatcher(c.facade.RawAPICaller(), result)
	return w, nil
}
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)
}