// WatchServices watches the services object on the api server
545236
func (c *APIClient) WatchServices(respCh chan SvcWatchResp) {
	ctx, _ := context.WithCancel(context.Background())

	go func() {
		// Make request to Kubernetes API
		getURL := c.watchBase + "services"
		req, err := http.NewRequest("GET", getURL, nil)
		if err != nil {
			respCh <- SvcWatchResp{opcode: "FATAL", errStr: fmt.Sprintf("Req %v", err)}
			return
		}
		if len(strings.TrimSpace(c.authToken)) > 0 {
			req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", c.authToken))
		}
		res, err := ctxhttp.Do(ctx, c.client, req)
		if err != nil {
			log.Errorf("Watch error: %v", err)
			respCh <- SvcWatchResp{opcode: "FATAL", errStr: fmt.Sprintf("Do %v", err)}
			return
		}
		defer res.Body.Close()

		var wss watchSvcStatus
		reader := bufio.NewReader(res.Body)

		// bufio.Reader.ReadBytes is blocking, so we watch for
		// context timeout or cancellation in a goroutine
		// and close the response body when see see it. The
		// response body is also closed via defer when the
		// request is made, but closing twice is OK.
		go func() {
			<-ctx.Done()
			res.Body.Close()
		}()

		for {
			line, err := reader.ReadBytes('\n')
			if ctx.Err() != nil {
				respCh <- SvcWatchResp{opcode: "ERROR", errStr: fmt.Sprintf("ctx %v", err)}
				return
			}
			if err != nil {
				respCh <- SvcWatchResp{opcode: "ERROR", errStr: fmt.Sprintf("read %v", err)}
				return
			}
			if err := json.Unmarshal(line, &wss); err != nil {
				respCh <- SvcWatchResp{opcode: "WARN", errStr: fmt.Sprintf("unmarshal %v", err)}
				continue
			}

			if wss.Object.ObjectMeta.Namespace == "kube-system" && (wss.Object.ObjectMeta.Name == "kube-scheduler" || wss.Object.ObjectMeta.Name == "kube-controller-manager") {
				// Ignoring these frequent updates
				continue
			}

			resp := SvcWatchResp{opcode: wss.Type}
			resp.svcName = wss.Object.ObjectMeta.Name
			sSpec := core.ServiceSpec{}
			sSpec.Ports = make([]core.PortSpec, 0, 1)
			sSpec.IPAddress = wss.Object.Spec.ClusterIP
			sSpec.ExternalIPs = wss.Object.Spec.ExternalIPs
			for _, port := range wss.Object.Spec.Ports {
				ps := core.PortSpec{Protocol: string(port.Protocol),
					SvcPort:  uint16(port.Port),
					NodePort: uint16(port.NodePort),
				}

				// handle 'kubernetes' service
				// Use port from configuration till named ports are supported.
				if resp.svcName == "kubernetes" && (len(wss.Object.ObjectMeta.Namespace) == 0 ||
					wss.Object.ObjectMeta.Namespace == "default") {
					ps.ProvPort = c.apiServerPort
				} else {
					ps.ProvPort = uint16(port.TargetPort)
				}
				sSpec.Ports = append(sSpec.Ports, ps)
			}

			resp.svcSpec = sSpec
			log.Infof("resp: %+v", resp)

			respCh <- resp
		}
	}()
}
1115266
func (st *State) watchEnqueuedActionsFilteredBy(receivers ...ActionReceiver) StringsWatcher {
	return newCollectionWatcher(st, colWCfg{
		col:    actionNotificationsC,
		filter: makeIdFilter(st, actionMarker, receivers...),
		idconv: actionNotificationIdToActionId,
	})
}
1271405
func (m *Machine) WatchUnits() (watcher.StringsWatcher, error) {
	var results params.StringsWatchResults
	args := params.Entities{
		Entities: []params.Entity{{Tag: m.tag.String()}},
	}
	err := m.facade.FacadeCall("WatchUnits", args, &results)
	if err != nil {
		return nil, err
	}
	if len(results.Results) != 1 {
		return nil, fmt.Errorf("expected 1 result, got %d", len(results.Results))
	}
	result := results.Results[0]
	if result.Error != nil {
		return nil, result.Error
	}
	w := apiwatcher.NewStringsWatcher(m.facade.RawAPICaller(), result)
	return w, nil
}
900558
func (facade *Facade) Watch() (params.StringsWatchResult, error) {
	watch := facade.backend.WatchScaledServices()
	if changes, ok := <-watch.Changes(); ok {
		id := facade.resources.Register(watch)
		return params.StringsWatchResult{
			StringsWatcherId: id,
			Changes:          changes,
		}, nil
	}
	return params.StringsWatchResult{}, watcher.EnsureErr(watch)
}
319323
func (a *API) WatchUnitAssignments() (params.StringsWatchResult, error) {
	watch := a.st.WatchForUnitAssignment()
	if changes, ok := <-watch.Changes(); ok {
		return params.StringsWatchResult{
			StringsWatcherId: a.res.Register(watch),
			Changes:          changes,
		}, nil
	}
	return params.StringsWatchResult{}, watcher.EnsureErr(watch)
}
360853
func (u *UniterAPIV8) WatchConfigSettings(args params.Entities) (params.NotifyWatchResults, error) {
	watcherFn := func(u *state.Unit) (state.NotifyWatcher, error) {
		return u.WatchConfigSettings()
	}
	result, err := u.WatchSettings(args, watcherFn)
	if err != nil {
		return params.NotifyWatchResults{}, errors.Trace(err)
	}

	return result, nil
}
535955
func (s *StorageProvisionerAPIv3) WatchVolumes(args params.Entities) (params.StringsWatchResults, error) {
	return s.watchStorageEntities(args, s.sb.WatchModelVolumes, s.sb.WatchMachineVolumes, nil)
}
157597
func (r *Request) Watch() (watch.Watch, error) {
	if r.err != nil {
		return nil, r.err
	}

	r.params.Set("watch", "true")

	req, err := r.request()
	if err != nil {
		return nil, err
	}

	w, err := watch.NewBodyWatcher(req, r.client)
	return w, err
}
748000
func (m *MockServiceAccountInterface) Watch(arg0 v11.ListOptions) (watch.Interface, error) {
	ret := m.ctrl.Call(m, "Watch", arg0)
	ret0, _ := ret[0].(watch.Interface)
	ret1, _ := ret[1].(error)
	return ret0, ret1
}
273854
func (w *checkWatcher) Watch(allocID, taskName, checkID string, check *structs.ServiceCheck, restarter TaskRestarter) {
	if !check.TriggersRestarts() {
		// Not watched, noop
		return
	}

	c := &checkRestart{
		allocID:        allocID,
		taskName:       taskName,
		checkID:        checkID,
		checkName:      check.Name,
		taskKey:        fmt.Sprintf("%s%s", allocID, taskName), // unique task ID
		task:           restarter,
		interval:       check.Interval,
		grace:          check.CheckRestart.Grace,
		graceUntil:     time.Now().Add(check.CheckRestart.Grace),
		timeLimit:      check.Interval * time.Duration(check.CheckRestart.Limit-1),
		ignoreWarnings: check.CheckRestart.IgnoreWarnings,
		logger:         w.logger.With("alloc_id", allocID, "task", taskName, "check", check.Name),
	}

	update := checkWatchUpdate{
		checkID:      checkID,
		checkRestart: c,
	}

	select {
	case w.checkUpdateCh <- update:
		// sent watch
	case <-w.done:
		// exited; nothing to do
	}
}
1101637
func (s *Store) parseServiceNodes(tx *memdb.Txn, ws memdb.WatchSet, services structs.ServiceNodes) (structs.ServiceNodes, error) {
	// We don't want to track an unlimited number of nodes, so we pull a
	// top-level watch to use as a fallback.
	allNodes, err := tx.Get("nodes", "id")
	if err != nil {
		return nil, fmt.Errorf("failed nodes lookup: %s", err)
	}
	allNodesCh := allNodes.WatchCh()

	// Fill in the node data for each service instance.
	var results structs.ServiceNodes
	for _, sn := range services {
		// Note that we have to clone here because we don't want to
		// modify the node-related fields on the object in the database,
		// which is what we are referencing.
		s := sn.PartialClone()

		// Grab the corresponding node record.
		watchCh, n, err := tx.FirstWatch("nodes", "id", sn.Node)
		if err != nil {
			return nil, fmt.Errorf("failed node lookup: %s", err)
		}
		ws.AddWithLimit(watchLimit, watchCh, allNodesCh)

		// Populate the node-related fields. The tagged addresses may be
		// used by agents to perform address translation if they are
		// configured to do that.
		node := n.(*structs.Node)
		s.ID = node.ID
		s.Address = node.Address
		s.Datacenter = node.Datacenter
		s.TaggedAddresses = node.TaggedAddresses
		s.NodeMeta = node.Meta

		results = append(results, s)
	}
	return results, nil
}
1197200
func (c *Client) WatchAll() (params.AllWatcherId, error) {
	if err := c.checkCanRead(); err != nil {
		return params.AllWatcherId{}, err
	}
	model, err := c.api.stateAccessor.Model()
	if err != nil {
		return params.AllWatcherId{}, errors.Trace(err)
	}

	// Since we know this is a user tag (because AuthClient is true),
	// we just do the type assertion to the UserTag.
	apiUser, _ := c.api.auth.GetAuthTag().(names.UserTag)
	isAdmin, err := common.HasModelAdmin(c.api.auth, apiUser, c.api.stateAccessor.ControllerTag(), model)
	if err != nil {
		return params.AllWatcherId{}, errors.Trace(err)
	}
	watchParams := state.WatchParams{IncludeOffers: isAdmin}

	w := c.api.stateAccessor.Watch(watchParams)
	return params.AllWatcherId{
		AllWatcherId: c.api.resources.Register(w),
	}, nil
}
179866
func (st *State) WatchModels() StringsWatcher {
	return newCollectionWatcher(st, colWCfg{
		col:    modelsC,
		global: true,
	})
}
661574
func (facade *Facade) Watch(entities params.Entities) params.NotifyWatchResults {
	count := len(entities.Entities)
	results := params.NotifyWatchResults{
		Results: make([]params.NotifyWatchResult, count),
	}
	for i, entity := range entities.Entities {
		id, err := facade.oneWatch(entity.Tag)
		results.Results[i].NotifyWatcherId = id
		results.Results[i].Error = common.ServerError(err)
	}
	return results
}
679189
func (u *UpgraderAPI) WatchAPIVersion(args params.Entities) (params.NotifyWatchResults, error) {
	result := params.NotifyWatchResults{
		Results: make([]params.NotifyWatchResult, len(args.Entities)),
	}
	for i, agent := range args.Entities {
		tag, err := names.ParseTag(agent.Tag)
		if err != nil {
			return params.NotifyWatchResults{}, errors.Trace(err)
		}
		err = common.ErrPerm
		if u.authorizer.AuthOwner(tag) {
			watch := u.m.WatchForModelConfigChanges()
			// Consume the initial event. Technically, API
			// calls to Watch 'transmit' the initial event
			// in the Watch response. But NotifyWatchers
			// have no state to transmit.
			if _, ok := <-watch.Changes(); ok {
				result.Results[i].NotifyWatcherId = u.resources.Register(watch)
				err = nil
			} else {
				err = watcher.EnsureErr(watch)
			}
		}
		result.Results[i].Error = common.ServerError(err)
	}
	return result, nil
}
218913
func (api *InstanceMutaterAPI) WatchContainers(arg params.Entity) (params.StringsWatchResult, error) {
	result := params.StringsWatchResult{}
	canAccess, err := api.getAuthFunc()
	if err != nil {
		return result, errors.Trace(err)
	}
	tag, err := names.ParseMachineTag(arg.Tag)
	if err != nil {
		return result, errors.Trace(err)
	}
	machine, err := api.getCacheMachine(canAccess, tag)
	if err != nil {
		return result, err
	}
	watch, err := machine.WatchContainers()
	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 machine containers")
	}
	return result, nil
}
852075
func (a *Application) Watch() NotifyWatcher {
	return newEntityWatcher(a.st, applicationsC, a.doc.DocID)
}