// rbdStatus checks if there is watcher on the image.
// It returns true if there is a watcher onthe image, otherwise returns false.
323905
func (u *RBDUtil) rbdStatus(image string, pOpts *rbdProvisionOptions) (bool, error) {
	var err error
	var output string
	var cmd []byte

	mon := u.kernelRBDMonitorsOpt(pOpts.monitors)
	// cmd "rbd status" list the rbd client watch with the following output:
	//
	// # there is a watcher (exit=0)
	// Watchers:
	//   watcher=10.16.153.105:0/710245699 client.14163 cookie=1
	//
	// # there is no watcher (exit=0)
	// Watchers: none
	//
	// Otherwise, exit is non-zero, for example:
	//
	// # image does not exist (exit=2)
	// rbd: error opening image kubernetes-dynamic-pvc-<UUID>: (2) No such file or directory
	//
	klog.V(4).Infof("rbd: status %s using mon %s, pool %s id %s key %s", image, mon, pOpts.pool, pOpts.adminID, pOpts.adminSecret)
	args := []string{"status", image, "--pool", pOpts.pool, "-m", mon, "--id", pOpts.adminID, "--key=" + pOpts.adminSecret}
	cmd, err = u.execCommand("rbd", args)
	output = string(cmd)

	// If command never succeed, returns its last error.
	if err != nil {
		return false, err
	}

	if strings.Contains(output, imageWatcherStr) {
		klog.V(4).Infof("rbd: watchers on %s: %s", image, output)
		return true, nil
	}
	klog.Warningf("rbd: no watchers on %s", image)
	return false, nil
}
284566
func (u *LXDProfileAPI) WatchLXDProfileUpgradeNotifications(args params.LXDProfileUpgrade) (params.StringsWatchResults, error) {
	u.logger.Tracef("Starting WatchLXDProfileUpgradeNotifications with %+v", args)
	result := params.StringsWatchResults{
		Results: make([]params.StringsWatchResult, len(args.Entities)),
	}
	canAccess, err := u.accessUnit()
	if err != nil {
		return params.StringsWatchResults{}, err
	}
	for i, entity := range args.Entities {
		tag, err := names.ParseTag(entity.Tag)
		if err != nil {
			result.Results[i].Error = common.ServerError(common.ErrPerm)
			continue
		}

		if !canAccess(tag) {
			result.Results[i].Error = common.ServerError(common.ErrPerm)
			continue
		}
		machine, err := u.getLXDProfileMachine(tag)
		if err != nil {
			result.Results[i].Error = common.ServerError(err)
			continue
		}

		watcherId, initial, err := u.watchOneChangeLXDProfileUpgradeNotifications(machine, args.ApplicationName)
		if err != nil {
			result.Results[i].Error = common.ServerError(err)
			continue
		}

		result.Results[i].StringsWatcherId = watcherId
		result.Results[i].Changes = initial
	}
	return result, nil
}
929932
func (api *CleanerAPI) WatchCleanups() (params.NotifyWatchResult, error) {
	watch := api.st.WatchCleanups()
	if _, ok := <-watch.Changes(); ok {
		return params.NotifyWatchResult{
			NotifyWatcherId: api.resources.Register(watch),
		}, nil
	}
	return params.NotifyWatchResult{
		Error: common.ServerError(watcher.EnsureErr(watch)),
	}, nil
}
972970
func (c *Client) WatchOfferStatus(arg params.OfferArg) (watcher.OfferStatusWatcher, error) {
	args := params.OfferArgs{Args: []params.OfferArg{arg}}
	// Use any previously cached discharge macaroons.
	if ms, ok := c.getCachedMacaroon("watch offer status", arg.OfferUUID); ok {
		args.Args[0].Macaroons = ms
	}

	var results params.OfferStatusWatchResults
	apiCall := func() error {
		// Reset the results struct before each api call.
		results = params.OfferStatusWatchResults{}
		if err := c.facade.FacadeCall("WatchOfferStatus", args, &results); err != nil {
			return errors.Trace(err)
		}
		if len(results.Results) != 1 {
			return errors.Errorf("expected 1 result, got %d", len(results.Results))
		}
		return nil
	}

	// Make the api call the first time.
	if err := apiCall(); err != nil {
		return nil, errors.Trace(err)
	}

	// On error, possibly discharge the macaroon and retry.
	result := results.Results[0]
	if result.Error != nil {
		mac, err := c.handleError(result.Error)
		if err != nil {
			result.Error.Message = err.Error()
			return nil, result.Error
		}
		args.Args[0].Macaroons = mac
		c.cache.Upsert(args.Args[0].OfferUUID, mac)

		if err := apiCall(); err != nil {
			return nil, errors.Trace(err)
		}
		result = results.Results[0]
	}
	if result.Error != nil {
		return nil, result.Error
	}

	w := apiwatcher.NewOfferStatusWatcher(c.facade.RawAPICaller(), result)
	return w, nil
}
405725
func (_m *Store) WatchDeployStatus(ctx context.Context, appname string, entrypoint string, nodename string) chan *types.DeployStatus {
	ret := _m.Called(ctx, appname, entrypoint, nodename)

	var r0 chan *types.DeployStatus
	if rf, ok := ret.Get(0).(func(context.Context, string, string, string) chan *types.DeployStatus); ok {
		r0 = rf(ctx, appname, entrypoint, nodename)
	} else {
		if ret.Get(0) != nil {
			r0 = ret.Get(0).(chan *types.DeployStatus)
		}
	}

	return r0
}
442889
func newActionStatusWatcher(backend modelBackend, receivers []ActionReceiver, statusSet ...ActionStatus) StringsWatcher {
	watchLogger.Debugf("newActionStatusWatcher receivers:'%+v', statuses'%+v'", receivers, statusSet)
	w := &actionStatusWatcher{
		commonWatcher:  newCommonWatcher(backend),
		source:         make(chan watcher.Change),
		sink:           make(chan []string),
		receiverFilter: actionReceiverInCollectionOp(receivers...),
		statusFilter:   statusInCollectionOp(statusSet...),
	}

	w.tomb.Go(func() error {
		defer close(w.sink)
		return w.loop()
	})

	return w
}
757450
func (u *Unit) WatchMeterStatus() NotifyWatcher {
	return newDocWatcher(u.st, []docKey{
		{
			meterStatusC,
			u.st.docID(u.globalMeterStatusKey()),
		}, {
			meterStatusC,
			metricsManagerKey(u.st),
		},
	})
}
281863
func setStatus(fd syscall.Handle, status bool) error {
	var bytesReturned uint32
	rdbbuf := make([]byte, syscall.MAXIMUM_REPARSE_DATA_BUFFER_SIZE)
	code := []byte{0x00, 0x00, 0x00, 0x00}
	if status {
		code[0] = 0x01
	}
	return syscall.DeviceIoControl(fd, tap_ioctl_set_media_status, &code[0], uint32(4), &rdbbuf[0], uint32(len(rdbbuf)), &bytesReturned, nil)
}
1057091
func (u *Unit) WatchLXDProfileUpgradeNotifications() (StringsWatcher, error) {
	app, err := u.Application()
	if err != nil {
		return nil, errors.Trace(err)
	}
	watchDocId := app.doc.DocID
	return watchInstanceCharmProfileCompatibilityData(u.st, watchDocId), nil
}
392596
func (s *Server) ImageStatus(ctx context.Context, req *pb.ImageStatusRequest) (resp *pb.ImageStatusResponse, err error) {
	const operation = "image_status"
	defer func() {
		recordOperation(operation, time.Now())
		recordError(operation, err)
	}()

	logrus.Debugf("ImageStatusRequest: %+v", req)
	image := ""
	img := req.GetImage()
	if img != nil {
		image = img.Image
	}
	if image == "" {
		return nil, fmt.Errorf("no image specified")
	}
	images, err := s.StorageImageServer().ResolveNames(image)
	if err != nil {
		if err == pkgstorage.ErrCannotParseImageID {
			images = append(images, image)
		} else {
			return nil, err
		}
	}
	var (
		notfound bool
		lastErr  error
	)
	for _, image := range images {
		status, err := s.StorageImageServer().ImageStatus(s.ImageContext(), image)
		if err != nil {
			if errors.Cause(err) == storage.ErrImageUnknown {
				logrus.Warnf("imageStatus: can't find %s", image)
				notfound = true
				continue
			}
			logrus.Warnf("imageStatus: error getting status from %s: %v", image, err)
			lastErr = err
			continue
		}

		// Ensure that size is already defined
		var size uint64
		if status.Size == nil {
			size = 0
		} else {
			size = *status.Size
		}

		resp = &pb.ImageStatusResponse{
			Image: &pb.Image{
				Id:          status.ID,
				RepoTags:    status.RepoTags,
				RepoDigests: status.RepoDigests,
				Size_:       size,
			},
		}
		uid, username := getUserFromImage(status.User)
		if uid != nil {
			resp.Image.Uid = &pb.Int64Value{Value: *uid}
		}
		resp.Image.Username = username
		break
	}
	if lastErr != nil && resp == nil {
		return nil, lastErr
	}
	if notfound && resp == nil {
		return &pb.ImageStatusResponse{}, nil
	}
	logrus.Debugf("ImageStatusResponse: %+v", resp)
	return resp, nil
}
475882
func (rbds *redisBlobDescriptorService) Stat(ctx context.Context, dgst digest.Digest) (distribution.Descriptor, error) {
	if err := dgst.Validate(); err != nil {
		return distribution.Descriptor{}, err
	}

	conn := rbds.pool.Get()
	defer conn.Close()

	return rbds.stat(ctx, conn, dgst)
}
138836
func (d *ConsulStateDriver) WatchAllState(baseKey string, sType core.State,
	unmarshal func([]byte, interface{}) error, rsps chan core.WatchState) error {
	baseKey = processKey(baseKey)
	byteRsps := make(chan [2][]byte, 1)
	recvErr := make(chan error, 1)

	go channelStateEvents(d, sType, unmarshal, byteRsps, rsps, recvErr)

	err := d.WatchAll(baseKey, byteRsps)
	if err != nil {
		return err
	}

	err = <-recvErr
	return err

}
1351739
func (keys *watchBrokerKeys) watchResync(resyncReg resync.Registration) {
	for resyncStatus := range resyncReg.StatusChan() {
		if resyncStatus.ResyncStatus() == resync.Started {
			err := keys.resync()
			if err != nil {
				// We are not able to propagate it somewhere else.
				logrus.DefaultLogger().Errorf("getting resync data failed: %v", err)
				// TODO NICE-to-HAVE publish the err using the transport asynchronously
			}
		}
		resyncStatus.Ack()
	}
}
329063
func (daemon *Daemon) newStatsCollector(interval time.Duration) *stats.Collector {
	// FIXME(vdemeester) move this elsewhere
	if runtime.GOOS == "linux" {
		meminfo, err := system.ReadMemInfo()
		if err == nil && meminfo.MemTotal > 0 {
			daemon.machineMemory = uint64(meminfo.MemTotal)
		}
	}
	s := stats.NewCollector(daemon, interval)
	go s.Run()
	return s
}
286629
func (c *Client) rpcRetryWatcher() <-chan struct{} {
	c.rpcRetryLock.Lock()
	defer c.rpcRetryLock.Unlock()

	if c.rpcRetryCh == nil {
		c.rpcRetryCh = make(chan struct{})
	}

	return c.rpcRetryCh
}
98118
func (st *state) WatchForRebootEvent() (watcher.NotifyWatcher, error) {
	var result params.NotifyWatchResult

	if err := st.facade.FacadeCall("WatchForRebootEvent", nil, &result); err != nil {
		return nil, err
	}
	if result.Error != nil {
		return nil, result.Error
	}

	w := apiwatcher.NewNotifyWatcher(st.facade.RawAPICaller(), result)
	return w, nil
}
409509
func (c *ConnectCALeaf) rootWatcher(ctx context.Context) {
	atomic.AddUint32(&c.testRootWatchStartCount, 1)
	defer atomic.AddUint32(&c.testRootWatchStopCount, 1)

	ch := make(chan cache.UpdateEvent, 1)
	err := c.Cache.Notify(ctx, ConnectCARootName, &structs.DCSpecificRequest{
		Datacenter: c.Datacenter,
	}, "roots", ch)

	notifyChange := func() {
		c.rootWatchMu.Lock()
		defer c.rootWatchMu.Unlock()

		for ch := range c.rootWatchSubscribers {
			select {
			case ch <- struct{}{}:
			default:
				// Don't block - chans are 1-buffered so act as an edge trigger and
				// reload CA state directly from cache so they never "miss" updates.
			}
		}
	}

	if err != nil {
		// Trigger all inflight watchers. We don't pass the error, but they will
		// reload from cache and observe the same error and return it to the caller,
		// or if it's transient, will continue and the next Fetch will get us back
		// into the right state. Seems better than busy loop-retrying here given
		// that almost any error we would see here would also be returned from the
		// cache get this will trigger.
		notifyChange()
		return
	}

	var oldRoots *structs.IndexedCARoots
	// Wait for updates to roots or all requests to stop
	for {
		select {
		case <-ctx.Done():
			return
		case e := <-ch:
			// Root response changed in some way. Note this might be the initial
			// fetch.
			if e.Err != nil {
				// See above rationale about the error propagation
				notifyChange()
				continue
			}

			roots, ok := e.Result.(*structs.IndexedCARoots)
			if !ok {
				// See above rationale about the error propagation
				notifyChange()
				continue
			}

			// Check that the active root is actually different from the last CA
			// config there are many reasons the config might have changed without
			// actually updating the CA root that is signing certs in the cluster.
			// The Fetch calls will also validate this since the first call here we
			// don't know if it changed or not, but there is no point waking up all
			// Fetch calls to check this if we know none of them will need to act on
			// this update.
			if oldRoots != nil && oldRoots.ActiveRootID == roots.ActiveRootID {
				continue
			}

			// Distribute the update to all inflight requests - they will decide
			// whether or not they need to act on it.
			notifyChange()
			oldRoots = roots
		}
	}
}