// SyncRegistry prompts the master to repush all images in the index into the
// docker registry.
329672
func (s *Server) SyncRegistry(req struct{}, reply *int) error {
	return s.f.SyncRegistryImages(s.context(), true)
}
1081925
func SetRegistryImage(conn client.Connection, rImage registry.Image) error {
	leaderpath := path.Join(zkregistryrepos, rImage.Library, rImage.Repo)
	leadernode := &RegistryImageLeader{HostID: "master"}
	if err := conn.CreateDir(leaderpath); err != nil && err != client.ErrNodeExists {
		glog.Errorf("Could not create repo path %s: %s", leaderpath, err)
		return err
	}
	imagepath := path.Join(zkregistrytags, rImage.ID())
	node := &RegistryImageNode{Image: rImage, PushedAt: time.Unix(0, 0)}
	if err := conn.Create(imagepath, node); err == client.ErrNodeExists {
		leader, err := conn.NewLeader(leaderpath)
		if err != nil {
			glog.Errorf("Could not establish leader at path %s: %s", leaderpath, err)
			return err
		}
		leaderDone := make(chan struct{})
		defer close(leaderDone)
		_, err = leader.TakeLead(leadernode, leaderDone)
		if err != nil {
			return err
		}
		defer leader.ReleaseLead()
		if err := conn.Get(imagepath, node); err != nil {
			return err
		}
		node.Image = rImage
		node.PushedAt = time.Unix(0, 0)
		return conn.Set(imagepath, node)
	} else if err != nil {
		glog.Errorf("Could not create tag path %s: %s", imagepath, err)
		return err
	}
	return nil

}
874890
func Register(asw cache.ActualStateOfWorld, dsw cache.DesiredStateOfWorld, pluginMgr *volume.VolumePluginMgr) {
	registerMetrics.Do(func() {
		prometheus.MustRegister(&totalVolumesCollector{asw, dsw, pluginMgr})
	})
}
367404
func (r *RegistryMux) ConnectToRegistry(e *engine.Engine) {
	for {
		// We have to check if the leader has changed to etcd otherwise keep grpc connection
		isGrpc, err := e.IsGrpcLeader()
		// If there is not error then we are able to get the leader state and continue
		// otherwise we have to wait
		if err == nil {
			if isGrpc {
				if r.rpcRegistry != nil && r.rpcRegistry.IsRegistryReady() {
					log.Infof("Reusing gRPC engine, connection is READY\n")
					r.currentRegistry = r.rpcRegistry
				} else {
					if r.rpcRegistry != nil {
						r.rpcRegistry.Close()
					}
					log.Infof("New engine supports gRPC, connecting\n")
					r.rpcRegistry = NewRPCRegistry(r.rpcDialerNoEngine)
					// connect to rpc registry
					r.rpcRegistry.Connect()
					r.currentRegistry = r.rpcRegistry
				}
			} else {
				if r.rpcRegistry != nil {
					r.rpcRegistry.Close()
				}
				// new leader is etcd-based
				r.currentRegistry = r.etcdRegistry
			}
		}
		time.Sleep(5 * time.Second)
	}
}
1218854
func (srv *Instance) RegisterRegistrationProvider(ctx context.Context, wg *sync.WaitGroup, provider registration.RegistrationDataProvider) error {
	return srv.registration.RegisterProvider(ctx, wg, provider)
}
238260
func (a *api) UpgradeRegistry(endpoint string, override bool) error {
	client, err := a.connectMaster()
	if err != nil {
		return err
	}
	return client.UpgradeRegistry(endpoint, override)
}
1144026
func (c *ServicedCli) cmdMigrateRegistry(ctx *cli.Context) {
	endpoint := ctx.String("registry")
	override := ctx.Bool("override")
	if err := c.driver.UpgradeRegistry(endpoint, override); err != nil {
		log.WithFields(logrus.Fields{
			"registry": endpoint,
			"override": override,
		}).WithError(err).Fatal("Unable to upgrade local Docker registry")
	}
}
1257957
func (daemon *Daemon) reloadInsecureRegistries(conf *config.Config, attributes map[string]string) error {
	// update corresponding configuration
	if conf.IsValueSet("insecure-registries") {
		daemon.configStore.InsecureRegistries = conf.InsecureRegistries
		if err := daemon.RegistryService.LoadInsecureRegistries(conf.InsecureRegistries); err != nil {
			return err
		}
	}

	// prepare reload event attributes with updatable configurations
	if daemon.configStore.InsecureRegistries != nil {
		insecureRegistries, err := json.Marshal(daemon.configStore.InsecureRegistries)
		if err != nil {
			return err
		}
		attributes["insecure-registries"] = string(insecureRegistries)
	} else {
		attributes["insecure-registries"] = "[]"
	}

	return nil
}
481530
func (d *Downloader) RegisterPeer(id string, version int, peer Peer) error {
	logger := log.New("peer", id)
	logger.Trace("Registering sync peer")
	if err := d.peers.Register(newPeerConnection(id, version, peer, logger)); err != nil {
		logger.Error("Failed to register sync peer", "err", err)
		return err
	}
	d.qosReduceConfidence()

	return nil
}
1124102
func (r *Reflector) syncWith(items []runtime.Object, resourceVersion string) error {
	found := make([]interface{}, 0, len(items))
	for _, item := range items {
		found = append(found, item)
	}
	return r.store.Replace(found, resourceVersion)
}
1018550
func (d *stiDocker) PullImage(name string) (*api.Image, error) {
	name = getImageName(name)

	// RegistryAuth is the base64 encoded credentials for the registry
	base64Auth, err := base64EncodeAuth(d.pullAuth)
	if err != nil {
		return nil, s2ierr.NewPullImageError(name, err)
	}
	var retriableError = false

	for retries := 0; retries <= DefaultPullRetryCount; retries++ {
		err = util.TimeoutAfter(DefaultDockerTimeout, fmt.Sprintf("pulling image %q", name), func(timer *time.Timer) error {
			resp, pullErr := d.client.ImagePull(context.Background(), name, dockertypes.ImagePullOptions{RegistryAuth: base64Auth})
			if pullErr != nil {
				return pullErr
			}
			defer resp.Close()

			decoder := json.NewDecoder(resp)
			for {
				if !timer.Stop() {
					return &util.TimeoutError{}
				}
				timer.Reset(DefaultDockerTimeout)

				var msg dockermessage.JSONMessage
				pullErr = decoder.Decode(&msg)
				if pullErr == io.EOF {
					return nil
				}
				if pullErr != nil {
					return pullErr
				}

				if msg.Error != nil {
					return msg.Error
				}
				if msg.Progress != nil {
					glog.V(4).Infof("pulling image %s: %s", name, msg.Progress.String())
				}
			}
		})
		if err == nil {
			break
		}
		glog.V(0).Infof("pulling image error : %v", err)
		errMsg := fmt.Sprintf("%s", err)
		for _, errorString := range RetriableErrors {
			if strings.Contains(errMsg, errorString) {
				retriableError = true
				break
			}
		}

		if !retriableError {
			return nil, s2ierr.NewPullImageError(name, err)
		}

		glog.V(0).Infof("retrying in %s ...", DefaultPullRetryDelay)
		time.Sleep(DefaultPullRetryDelay)
	}

	inspectResp, err := d.InspectImage(name)
	if err != nil {
		return nil, s2ierr.NewPullImageError(name, err)
	}
	if inspectResp != nil {
		image := &api.Image{}
		updateImageWithInspect(image, inspectResp)
		return image, nil
	}
	return nil, nil
}
786373
func IsRegistryDockerHub(registry string) bool {
	switch registry {
	case DockerDefaultRegistry, DockerDefaultV1Registry, DockerDefaultV2Registry:
		return true
	default:
		return false
	}
}
19242
func GetRegistryImage(conn client.Connection, id string) (*registry.Image, error) {
	rimagepath := path.Join(zkregistrytags, id)
	var node RegistryImageNode
	if err := conn.Get(rimagepath, &node); err != nil {
		return nil, err
	}
	return &node.Image, nil
}
922894
func (s *SeesawSync) Register(node net.IP, id *SyncSessionID) error {
	if id == nil {
		return errors.New("id is nil")
	}

	// TODO(jsing): Reject if not master?

	s.sync.sessionLock.Lock()
	session := newSyncSession(node, s.sync.nextSessionID)
	s.sync.nextSessionID++
	s.sync.sessions[session.id] = session
	s.sync.sessionLock.Unlock()

	session.Lock()
	session.expiryTime = time.Now().Add(sessionDeadtime)
	session.Unlock()

	*id = session.id

	log.Infof("Synchronisation session %d registered by %v", *id, node)

	return nil
}
978608
func NewRegistry(applyer Applyer) *Registry {
	return &Registry{
		applyer:     applyer,
		sources:     make(map[string]Source),
		schemas:     make(map[string]*Schema),
		schemaNames: make([]string, 0),
	}
}
299543
func NewRegistry(localID enode.ID, delivery *Delivery, syncChunkStore storage.SyncChunkStore, intervalsStore state.Store, options *RegistryOptions, balance protocols.Balance) *Registry {
	if options == nil {
		options = &RegistryOptions{}
	}
	if options.SyncUpdateDelay <= 0 {
		options.SyncUpdateDelay = 15 * time.Second
	}
	// check if retrieval has been disabled
	retrieval := options.Retrieval != RetrievalDisabled

	quit := make(chan struct{})

	streamer := &Registry{
		addr:           localID,
		skipCheck:      options.SkipCheck,
		serverFuncs:    make(map[string]func(*Peer, string, bool) (Server, error)),
		clientFuncs:    make(map[string]func(*Peer, string, bool) (Client, error)),
		peers:          make(map[enode.ID]*Peer),
		delivery:       delivery,
		intervalsStore: intervalsStore,
		autoRetrieval:  retrieval,
		maxPeerServers: options.MaxPeerServers,
		balance:        balance,
		quit:           quit,
	}

	streamer.setupSpec()

	streamer.api = NewAPI(streamer)
	delivery.getPeer = streamer.getPeer

	// if retrieval is enabled, register the server func, so that retrieve requests will be served (non-light nodes only)
	if options.Retrieval == RetrievalEnabled {
		streamer.RegisterServerFunc(swarmChunkServerStreamName, func(_ *Peer, _ string, live bool) (Server, error) {
			if !live {
				return nil, errors.New("only live retrieval requests supported")
			}
			return NewSwarmChunkServer(delivery.chunkStore), nil
		})
	}

	// if retrieval is not disabled, register the client func (both light nodes and normal nodes can issue retrieve requests)
	if options.Retrieval != RetrievalDisabled {
		streamer.RegisterClientFunc(swarmChunkServerStreamName, func(p *Peer, t string, live bool) (Client, error) {
			return NewSwarmSyncerClient(p, syncChunkStore, NewStream(swarmChunkServerStreamName, t, live))
		})
	}

	// If syncing is not disabled, the syncing functions are registered (both client and server)
	if options.Syncing != SyncingDisabled {
		RegisterSwarmSyncerServer(streamer, syncChunkStore)
		RegisterSwarmSyncerClient(streamer, syncChunkStore)
	}

	// if syncing is set to automatically subscribe to the syncing stream, start the subscription process
	if options.Syncing == SyncingAutoSubscribe {
		// latestIntC function ensures that
		//   - receiving from the in chan is not blocked by processing inside the for loop
		// 	 - the latest int value is delivered to the loop after the processing is done
		// In context of NeighbourhoodDepthC:
		// after the syncing is done updating inside the loop, we do not need to update on the intermediate
		// depth changes, only to the latest one
		latestIntC := func(in <-chan int) <-chan int {
			out := make(chan int, 1)

			go func() {
				defer close(out)

				for {
					select {
					case i, ok := <-in:
						if !ok {
							return
						}
						select {
						case <-out:
						default:
						}
						out <- i
					case <-quit:
						return
					}
				}
			}()

			return out
		}

		kad := streamer.delivery.kad
		// get notification channels from Kademlia before returning
		// from this function to avoid race with Close method and
		// the goroutine created below
		depthC := latestIntC(kad.NeighbourhoodDepthC())
		addressBookSizeC := latestIntC(kad.AddrCountC())

		go func() {
			// wait for kademlia table to be healthy
			// but return if Registry is closed before
			select {
			case <-time.After(options.SyncUpdateDelay):
			case <-quit:
				return
			}

			// initial requests for syncing subscription to peers
			streamer.updateSyncing()

			for depth := range depthC {
				log.Debug("Kademlia neighbourhood depth change", "depth", depth)

				// Prevent too early sync subscriptions by waiting until there are no
				// new peers connecting. Sync streams updating will be done after no
				// peers are connected for at least SyncUpdateDelay period.
				timer := time.NewTimer(options.SyncUpdateDelay)
				// Hard limit to sync update delay, preventing long delays
				// on a very dynamic network
				maxTimer := time.NewTimer(3 * time.Minute)
			loop:
				for {
					select {
					case <-maxTimer.C:
						// force syncing update when a hard timeout is reached
						log.Trace("Sync subscriptions update on hard timeout")
						// request for syncing subscription to new peers
						streamer.updateSyncing()
						break loop
					case <-timer.C:
						// start syncing as no new peers has been added to kademlia
						// for some time
						log.Trace("Sync subscriptions update")
						// request for syncing subscription to new peers
						streamer.updateSyncing()
						break loop
					case size := <-addressBookSizeC:
						log.Trace("Kademlia address book size changed on depth change", "size", size)
						// new peers has been added to kademlia,
						// reset the timer to prevent early sync subscriptions
						if !timer.Stop() {
							<-timer.C
						}
						timer.Reset(options.SyncUpdateDelay)
					case <-quit:
						break loop
					}
				}
				timer.Stop()
				maxTimer.Stop()
			}
		}()
	}

	return streamer
}
1060408
func (c *Client) UpdateRegistration(accountKey interface{}, reg Registration) (Registration, error) {
	url := c.resources.Registration + strconv.Itoa(reg.Id)
	return c.registration(accountKey, reg, resourceRegistration, url)
}