// syncService will sync the Service with the given key if it has had its expectations fulfilled,
// meaning it did not expect to see any more of its pods created or deleted. This function is not meant to be
// invoked concurrently with the same key.
438092
func (s *ServiceController) syncService(key string) error {
	startTime := time.Now()
	var cachedService *cachedService
	defer func() {
		klog.V(4).Infof("Finished syncing service %q (%v)", key, time.Since(startTime))
	}()

	namespace, name, err := cache.SplitMetaNamespaceKey(key)
	if err != nil {
		return err
	}

	// service holds the latest service info from apiserver
	service, err := s.serviceLister.Services(namespace).Get(name)
	switch {
	case errors.IsNotFound(err):
		// service absence in store means watcher caught the deletion, ensure LB info is cleaned
		klog.Infof("Service has been deleted %v. Attempting to cleanup load balancer resources", key)
		err = s.processServiceDeletion(key)
	case err != nil:
		klog.Infof("Unable to retrieve service %v from store: %v", key, err)
	default:
		cachedService = s.cache.getOrCreate(key)
		err = s.processServiceUpdate(cachedService, service, key)
	}

	return err
}
1025475
func (pm *ProtocolManager) syncer() {
	// Start and ensure cleanup of sync mechanisms
	//pm.fetcher.Start()
	//defer pm.fetcher.Stop()
	defer pm.downloader.Terminate()

	// Wait for different events to fire synchronisation operations
	//forceSync := time.Tick(forceSyncCycle)
	for {
		select {
		case <-pm.newPeerCh:
			/*			// Make sure we have peers to select from, then sync
						if pm.peers.Len() < minDesiredPeerCount {
							break
						}
						go pm.synchronise(pm.peers.BestPeer())
			*/
		/*case <-forceSync:
		// Force a sync even if not enough peers are present
		go pm.synchronise(pm.peers.BestPeer())
		*/
		case <-pm.noMorePeers:
			return
		}
	}
}
1225045
func (p *Pod) Sync() error {
	cfd, err := p.Fd()
	if err != nil {
		return errwrap.Wrap(fmt.Errorf("error acquiring pod %v dir fd", p.UUID.String()), err)
	}
	if err := sys.Syncfs(cfd); err != nil {
		return errwrap.Wrap(fmt.Errorf("failed to sync pod %v data", p.UUID.String()), err)
	}
	return nil
}
1260998
func (k *kubernetesClient) deleteService(deploymentName string) error {
	services := k.client().CoreV1().Services(k.namespace)
	err := services.Delete(deploymentName, &v1.DeleteOptions{
		PropagationPolicy: &defaultPropagationPolicy,
	})
	if k8serrors.IsNotFound(err) {
		return nil
	}
	return errors.Trace(err)
}
977121
func (s *serviceCache) allServices() []*v1.Service {
	s.mu.Lock()
	defer s.mu.Unlock()
	services := make([]*v1.Service, 0, len(s.serviceMap))
	for _, v := range s.serviceMap {
		services = append(services, v.state)
	}
	return services
}
217109
func (l *State) syncService(id string) error {
	// If the service has associated checks that are out of sync,
	// piggyback them on the service sync so they are part of the
	// same transaction and are registered atomically. We only let
	// checks ride on service registrations with the same token,
	// otherwise we need to register them separately so they don't
	// pick up privileges from the service token.
	var checks structs.HealthChecks
	for checkID, c := range l.checks {
		if c.Deleted || c.InSync {
			continue
		}
		if c.Check.ServiceID != id {
			continue
		}
		if l.serviceToken(id) != l.checkToken(checkID) {
			continue
		}
		checks = append(checks, c.Check)
	}

	req := structs.RegisterRequest{
		Datacenter:      l.config.Datacenter,
		ID:              l.config.NodeID,
		Node:            l.config.NodeName,
		Address:         l.config.AdvertiseAddr,
		TaggedAddresses: l.config.TaggedAddresses,
		NodeMeta:        l.metadata,
		Service:         l.services[id].Service,
		WriteRequest:    structs.WriteRequest{Token: l.serviceToken(id)},
	}

	// Backwards-compatibility for Consul < 0.5
	if len(checks) == 1 {
		req.Check = checks[0]
	} else {
		req.Checks = checks
	}

	var out struct{}
	err := l.Delegate.RPC("Catalog.Register", &req, &out)
	switch {
	case err == nil:
		l.services[id].InSync = true
		// Given how the register API works, this info is also updated
		// every time we sync a service.
		l.nodeInfoInSync = true
		for _, check := range checks {
			l.checks[check.CheckID].InSync = true
		}
		l.logger.Printf("[INFO] agent: Synced service %q", id)
		return nil

	case acl.IsErrPermissionDenied(err), acl.IsErrNotFound(err):
		// todo(fs): mark the service and the checks to be in sync to prevent excessive retrying before next full sync
		// todo(fs): some backoff strategy might be a better solution
		l.services[id].InSync = true
		for _, check := range checks {
			l.checks[check.CheckID].InSync = true
		}
		l.logger.Printf("[WARN] agent: Service %q registration blocked by ACLs", id)
		metrics.IncrCounter([]string{"acl", "blocked", "service", "registration"}, 1)
		return nil

	default:
		l.logger.Printf("[WARN] agent: Syncing service %q failed. %s", id, err)
		return err
	}
}
887780
func (c *ServiceClient) sync() error {
	sreg, creg, sdereg, cdereg := 0, 0, 0, 0

	consulServices, err := c.client.Services()
	if err != nil {
		metrics.IncrCounter([]string{"client", "consul", "sync_failure"}, 1)
		return fmt.Errorf("error querying Consul services: %v", err)
	}

	consulChecks, err := c.client.Checks()
	if err != nil {
		metrics.IncrCounter([]string{"client", "consul", "sync_failure"}, 1)
		return fmt.Errorf("error querying Consul checks: %v", err)
	}

	// Remove Nomad services in Consul but unknown locally
	for id := range consulServices {
		if _, ok := c.services[id]; ok {
			// Known service, skip
			continue
		}

		// Ignore if this is not a Nomad managed service. Also ignore
		// Nomad managed services if this is not a client agent.
		// This is to prevent server agents from removing services
		// registered by client agents
		if !isNomadService(id) || !c.isClientAgent {
			// Not managed by Nomad, skip
			continue
		}

		// Unknown Nomad managed service; kill
		if err := c.client.ServiceDeregister(id); err != nil {
			if isOldNomadService(id) {
				// Don't hard-fail on old entries. See #3620
				continue
			}

			metrics.IncrCounter([]string{"client", "consul", "sync_failure"}, 1)
			return err
		}
		sdereg++
		metrics.IncrCounter([]string{"client", "consul", "service_deregistrations"}, 1)
	}

	// Add Nomad services missing from Consul
	for id, locals := range c.services {
		if _, ok := consulServices[id]; !ok {
			if err = c.client.ServiceRegister(locals); err != nil {
				metrics.IncrCounter([]string{"client", "consul", "sync_failure"}, 1)
				return err
			}
			sreg++
			metrics.IncrCounter([]string{"client", "consul", "service_registrations"}, 1)
		}
	}

	// Remove Nomad checks in Consul but unknown locally
	for id, check := range consulChecks {
		if _, ok := c.checks[id]; ok {
			// Known check, leave it
			continue
		}

		// Ignore if this is not a Nomad managed check. Also ignore
		// Nomad managed checks if this is not a client agent.
		// This is to prevent server agents from removing checks
		// registered by client agents
		if !isNomadService(check.ServiceID) || !c.isClientAgent {
			// Service not managed by Nomad, skip
			continue
		}

		// Unknown Nomad managed check; remove
		if err := c.client.CheckDeregister(id); err != nil {
			if isOldNomadService(check.ServiceID) {
				// Don't hard-fail on old entries.
				continue
			}

			metrics.IncrCounter([]string{"client", "consul", "sync_failure"}, 1)
			return err
		}
		cdereg++
		metrics.IncrCounter([]string{"client", "consul", "check_deregistrations"}, 1)
	}

	// Add Nomad checks missing from Consul
	for id, check := range c.checks {
		if _, ok := consulChecks[id]; ok {
			// Already in Consul; skipping
			continue
		}

		if err := c.client.CheckRegister(check); err != nil {
			metrics.IncrCounter([]string{"client", "consul", "sync_failure"}, 1)
			return err
		}
		creg++
		metrics.IncrCounter([]string{"client", "consul", "check_registrations"}, 1)

		// Handle starting scripts
		if script, ok := c.scripts[id]; ok {
			// If it's already running, cancel and replace
			if oldScript, running := c.runningScripts[id]; running {
				oldScript.cancel()
			}
			// Start and store the handle
			c.runningScripts[id] = script.run()
		}
	}

	// Only log if something was actually synced
	if sreg > 0 || sdereg > 0 || creg > 0 || cdereg > 0 {
		c.logger.Debug("sync complete", "registered_services", sreg, "deregistered_services", sdereg,
			"registered_checks", creg, "deregistered_checks", cdereg)
	}
	return nil
}
1019339
func (k *kubernetesClient) EnsureService(
	appName string,
	statusCallback caas.StatusCallbackFunc,
	params *caas.ServiceParams,
	numUnits int,
	config application.ConfigAttributes,
) (err error) {
	defer func() {
		if err != nil {
			_ = statusCallback(appName, status.Error, err.Error(), nil)
		}
	}()

	logger.Debugf("creating/updating application %s", appName)
	deploymentName := k.deploymentName(appName)

	if numUnits < 0 {
		return errors.Errorf("number of units must be >= 0")
	}
	if numUnits == 0 {
		return k.deleteAllPods(appName, deploymentName)
	}
	if params == nil || params.PodSpec == nil {
		return errors.Errorf("missing pod spec")
	}
	if params.PodSpec.OmitServiceFrontend && len(params.Filesystems) == 0 {
		return errors.Errorf("kubernetes service is required when using storage")
	}

	var cleanups []func()
	defer func() {
		if err == nil {
			return
		}
		for _, f := range cleanups {
			f()
		}
	}()

	unitSpec, err := makeUnitSpec(appName, deploymentName, params.PodSpec)
	if err != nil {
		return errors.Annotatef(err, "parsing unit spec for %s", appName)
	}
	if len(params.Devices) > 0 {
		if err = k.configureDevices(unitSpec, params.Devices); err != nil {
			return errors.Annotatef(err, "configuring devices for %s", appName)
		}
	}
	if mem := params.Constraints.Mem; mem != nil {
		if err = k.configureConstraint(unitSpec, "memory", fmt.Sprintf("%dMi", *mem)); err != nil {
			return errors.Annotatef(err, "configuring memory constraint for %s", appName)
		}
	}
	if cpu := params.Constraints.CpuPower; cpu != nil {
		if err = k.configureConstraint(unitSpec, "cpu", fmt.Sprintf("%dm", *cpu)); err != nil {
			return errors.Annotatef(err, "configuring cpu constraint for %s", appName)
		}
	}

	// Translate tags to node affinity.
	if params.Constraints.Tags != nil {
		affinityLabels := *params.Constraints.Tags
		var (
			affinityTags     = make(map[string]string)
			antiAffinityTags = make(map[string]string)
		)
		for _, labelPair := range affinityLabels {
			parts := strings.Split(labelPair, "=")
			if len(parts) != 2 {
				return errors.Errorf("invalid node affinity constraints: %v", affinityLabels)
			}
			key := strings.Trim(parts[0], " ")
			value := strings.Trim(parts[1], " ")
			if strings.HasPrefix(key, "^") {
				if len(key) == 1 {
					return errors.Errorf("invalid node affinity constraints: %v", affinityLabels)
				}
				antiAffinityTags[key[1:]] = value
			} else {
				affinityTags[key] = value
			}
		}

		updateSelectorTerms := func(nodeSelectorTerm *core.NodeSelectorTerm, tags map[string]string, op core.NodeSelectorOperator) {
			// Sort for stable ordering.
			var keys []string
			for k := range tags {
				keys = append(keys, k)
			}
			sort.Strings(keys)
			for _, tag := range keys {
				allValues := strings.Split(tags[tag], "|")
				for i, v := range allValues {
					allValues[i] = strings.Trim(v, " ")
				}
				nodeSelectorTerm.MatchExpressions = append(nodeSelectorTerm.MatchExpressions, core.NodeSelectorRequirement{
					Key:      tag,
					Operator: op,
					Values:   allValues,
				})
			}
		}
		var nodeSelectorTerm core.NodeSelectorTerm
		updateSelectorTerms(&nodeSelectorTerm, affinityTags, core.NodeSelectorOpIn)
		updateSelectorTerms(&nodeSelectorTerm, antiAffinityTags, core.NodeSelectorOpNotIn)
		unitSpec.Pod.Affinity = &core.Affinity{
			NodeAffinity: &core.NodeAffinity{
				RequiredDuringSchedulingIgnoredDuringExecution: &core.NodeSelector{
					NodeSelectorTerms: []core.NodeSelectorTerm{nodeSelectorTerm},
				},
			},
		}
	}
	if params.Constraints.Zones != nil {
		zones := *params.Constraints.Zones
		affinity := unitSpec.Pod.Affinity
		if affinity == nil {
			affinity = &core.Affinity{
				NodeAffinity: &core.NodeAffinity{
					RequiredDuringSchedulingIgnoredDuringExecution: &core.NodeSelector{
						NodeSelectorTerms: []core.NodeSelectorTerm{{}},
					},
				},
			}
			unitSpec.Pod.Affinity = affinity
		}
		nodeSelector := &affinity.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms[0]
		nodeSelector.MatchExpressions = append(nodeSelector.MatchExpressions,
			core.NodeSelectorRequirement{
				Key:      "failure-domain.beta.kubernetes.io/zone",
				Operator: core.NodeSelectorOpIn,
				Values:   zones,
			})
	}

	annotations := resourceTagsToAnnotations(params.ResourceTags)

	for _, c := range params.PodSpec.Containers {
		if c.ImageDetails.Password == "" {
			continue
		}
		imageSecretName := appSecretName(deploymentName, c.Name)
		if err := k.ensureOCIImageSecret(imageSecretName, appName, &c.ImageDetails, annotations.Copy()); err != nil {
			return errors.Annotatef(err, "creating secrets for container: %s", c.Name)
		}
		cleanups = append(cleanups, func() { k.deleteSecret(imageSecretName) })
	}
	// Add a deployment controller or stateful set configured to create the specified number of units/pods.
	// Defensively check to see if a stateful set is already used.
	var useStatefulSet bool
	if params.Deployment.ServiceType != "" {
		useStatefulSet = params.Deployment.DeploymentType == caas.DeploymentStateful
	} else {
		useStatefulSet = len(params.Filesystems) > 0
	}
	statefulsets := k.client().AppsV1().StatefulSets(k.namespace)
	existingStatefulSet, err := statefulsets.Get(deploymentName, v1.GetOptions{IncludeUninitialized: true})
	if err != nil && !k8serrors.IsNotFound(err) {
		return errors.Trace(err)
	}
	if !useStatefulSet {
		useStatefulSet = err == nil
		if useStatefulSet {
			logger.Debugf("no updated filesystems but already using stateful set for %v", appName)
		}
	}
	var randPrefix string
	if useStatefulSet {
		// Include a random snippet in the pvc name so that if the same app
		// is deleted and redeployed again, the pvc retains a unique name.
		// Only generate it once, and record it on the stateful set.
		if existingStatefulSet != nil {
			randPrefix = existingStatefulSet.Annotations[labelApplicationUUID]
		}
		if randPrefix == "" {
			var randPrefixBytes [4]byte
			if _, err := io.ReadFull(rand.Reader, randPrefixBytes[0:4]); err != nil {
				return errors.Trace(err)
			}
			randPrefix = fmt.Sprintf("%x", randPrefixBytes)
		}
	}

	numPods := int32(numUnits)
	if useStatefulSet {
		if err := k.configureStatefulSet(appName, deploymentName, randPrefix, annotations.Copy(), unitSpec, params.PodSpec.Containers, &numPods, params.Filesystems); err != nil {
			return errors.Annotate(err, "creating or updating StatefulSet")
		}
		cleanups = append(cleanups, func() { k.deleteDeployment(appName) })
	} else {
		if err := k.configureDeployment(appName, deploymentName, annotations.Copy(), unitSpec, params.PodSpec.Containers, &numPods); err != nil {
			return errors.Annotate(err, "creating or updating DeploymentController")
		}
		cleanups = append(cleanups, func() { k.deleteDeployment(appName) })
	}

	var ports []core.ContainerPort
	for _, c := range unitSpec.Pod.Containers {
		for _, p := range c.Ports {
			if p.ContainerPort == 0 {
				continue
			}
			ports = append(ports, p)
		}
	}
	if !params.PodSpec.OmitServiceFrontend {
		// Merge any service annotations from the charm.
		if unitSpec.Service != nil {
			annotations.Merge(k8sannotations.New(unitSpec.Service.Annotations))
		}
		// Merge any service annotations from the CLI.
		deployAnnotations, err := config.GetStringMap(serviceAnnotationsKey, nil)
		if err != nil {
			return errors.Annotatef(err, "unexpected annotations: %#v", config.Get(serviceAnnotationsKey, nil))
		}
		annotations.Merge(k8sannotations.New(deployAnnotations))

		config[serviceAnnotationsKey] = annotations.ToMap()
		if err := k.configureService(appName, deploymentName, ports, params, config); err != nil {
			return errors.Annotatef(err, "creating or updating service for %v", appName)
		}
	}
	return nil
}
784981
func (c *HyperClient) UpdateService(podID string, services []*types.UserService) error {
	_, err := c.client.ServiceUpdate(
		c.ctx,
		&types.ServiceUpdateRequest{PodID: podID, Services: services},
	)
	if err != nil {
		return err
	}

	return nil
}
830816
func (c *Controller) syncHandler(key string) error {
	// Convert the namespace/name string into a distinct namespace and name
	namespace, name, err := cache.SplitMetaNamespaceKey(key)
	if err != nil {
		utilruntime.HandleError(fmt.Errorf("invalid resource key: %s", key))
		return nil
	}

	// Get the Foo resource with this namespace/name
	foo, err := c.foosLister.Foos(namespace).Get(name)
	if err != nil {
		// The Foo resource may no longer exist, in which case we stop
		// processing.
		if errors.IsNotFound(err) {
			utilruntime.HandleError(fmt.Errorf("foo '%s' in work queue no longer exists", key))
			return nil
		}

		return err
	}

	deploymentName := foo.Spec.DeploymentName
	if deploymentName == "" {
		// We choose to absorb the error here as the worker would requeue the
		// resource otherwise. Instead, the next time the resource is updated
		// the resource will be queued again.
		utilruntime.HandleError(fmt.Errorf("%s: deployment name must be specified", key))
		return nil
	}

	// Get the deployment with the name specified in Foo.spec
	deployment, err := c.deploymentsLister.Deployments(foo.Namespace).Get(deploymentName)
	// If the resource doesn't exist, we'll create it
	if errors.IsNotFound(err) {
		deployment, err = c.kubeclientset.AppsV1().Deployments(foo.Namespace).Create(newDeployment(foo))
	}

	// If an error occurs during Get/Create, we'll requeue the item so we can
	// attempt processing again later. This could have been caused by a
	// temporary network failure, or any other transient reason.
	if err != nil {
		return err
	}

	// If the Deployment is not controlled by this Foo resource, we should log
	// a warning to the event recorder and ret
	if !metav1.IsControlledBy(deployment, foo) {
		msg := fmt.Sprintf(MessageResourceExists, deployment.Name)
		c.recorder.Event(foo, corev1.EventTypeWarning, ErrResourceExists, msg)
		return fmt.Errorf(msg)
	}

	// If this number of the replicas on the Foo resource is specified, and the
	// number does not equal the current desired replicas on the Deployment, we
	// should update the Deployment resource.
	if foo.Spec.Replicas != nil && *foo.Spec.Replicas != *deployment.Spec.Replicas {
		klog.V(4).Infof("Foo %s replicas: %d, deployment replicas: %d", name, *foo.Spec.Replicas, *deployment.Spec.Replicas)
		deployment, err = c.kubeclientset.AppsV1().Deployments(foo.Namespace).Update(newDeployment(foo))
	}

	// If an error occurs during Update, we'll requeue the item so we can
	// attempt processing again later. THis could have been caused by a
	// temporary network failure, or any other transient reason.
	if err != nil {
		return err
	}

	// Finally, we update the status block of the Foo resource to reflect the
	// current state of the world
	err = c.updateFooStatus(foo, deployment)
	if err != nil {
		return err
	}

	c.recorder.Event(foo, corev1.EventTypeNormal, SuccessSynced, MessageResourceSynced)
	return nil
}
780169
func (c *HyperClient) DeleteService(podID string, services []*types.UserService) error {
	_, err := c.client.ServiceDelete(
		c.ctx,
		&types.ServiceDelRequest{PodID: podID, Services: services},
	)

	if err != nil {
		return err
	}
	return nil
}
569497
func (ipvsc *ipvsControllerController) sync(key string) error {
	ipvsc.reloadRateLimiter.Accept()

	if !ipvsc.epController.HasSynced() || !ipvsc.svcController.HasSynced() {
		time.Sleep(100 * time.Millisecond)
		return fmt.Errorf("deferring sync till endpoints controller has synced")
	}

	ns, name, err := parseNsName(ipvsc.configMapName)
	if err != nil {
		glog.Warningf("%v", err)
		return err
	}
	cfgMap, err := ipvsc.getConfigMap(ns, name)
	if err != nil {
		return fmt.Errorf("unexpected error searching configmap %v: %v", ipvsc.configMapName, err)
	}

	svc := ipvsc.getServices(cfgMap)
	ipvsc.ruCfg = svc

	err = ipvsc.keepalived.WriteCfg(svc)
	if err != nil {
		return err
	}
	glog.V(2).Infof("services: %v", svc)

	md5, err := checksum(keepalivedCfg)
	if err == nil && md5 == ipvsc.ruMD5 {
		return nil
	}

	ipvsc.ruMD5 = md5
	err = ipvsc.keepalived.Reload()
	if err != nil {
		glog.Errorf("error reloading keepalived: %v", err)
	}

	return nil
}
72577
func (c *ConfigLocal) SetKeybaseService(k KeybaseService) {
	c.lock.Lock()
	defer c.lock.Unlock()
	c.service = k
}
576351
func (c *Client) UpdateService(i *UpdateServiceInput) (*Service, error) {
	if i.ID == "" {
		return nil, ErrMissingID
	}

	path := fmt.Sprintf("/service/%s", i.ID)
	resp, err := c.PutForm(path, i, nil)
	if err != nil {
		return nil, err
	}

	var s *Service
	if err := decodeJSON(&s, resp.Body); err != nil {
		return nil, err
	}
	return s, nil
}
738858
func (_m *API) UpdateService(_a0 io.Reader) (*service.ServiceDetails, error) {
	ret := _m.Called(_a0)

	var r0 *service.ServiceDetails
	if rf, ok := ret.Get(0).(func(io.Reader) *service.ServiceDetails); ok {
		r0 = rf(_a0)
	} else {
		if ret.Get(0) != nil {
			r0 = ret.Get(0).(*service.ServiceDetails)
		}
	}

	var r1 error
	if rf, ok := ret.Get(1).(func(io.Reader) error); ok {
		r1 = rf(_a0)
	} else {
		r1 = ret.Error(1)
	}

	return r0, r1
}
988952
func SyncServices(conn client.Connection, svcs []service.Service) error {
	pth := path.Join("/services")

	logger := plog.WithField("zkpath", pth)

	// look up children service ids
	ch, err := conn.Children(pth)
	if err != nil && err != client.ErrNoNode {
		logger.WithError(err).Debug("Could not look up services")

		// TODO: wrap error?
		return err
	}

	// store the service ids in a hash map
	chmap := make(map[string]struct{})
	for _, serviceid := range ch {
		chmap[serviceid] = struct{}{}
	}

	// set the services
	for _, s := range svcs {
		if err := UpdateService(conn, &s, false, false); err != nil {
			return err
		}

		// delete matching records
		if _, ok := chmap[s.ID]; ok {
			delete(chmap, s.ID)
		}
	}

	// remove any leftovers
	for serviceid := range chmap {
		if err := conn.Delete(path.Join(pth, serviceid)); err != nil {

			logger.WithField("serviceid", serviceid).WithError(err).Debug("Could not delete service entry from zookeeper")
			return &ServiceError{
				Action:    "sync",
				ServiceID: serviceid,
				Message:   "could not delete service",
			}
		}
	}

	return nil
}
316303
func (c *Client) SnapshotService(snapshotterName string) snapshots.Snapshotter {
	if c.snapshotters != nil {
		return c.snapshotters[snapshotterName]
	}
	c.connMu.Lock()
	defer c.connMu.Unlock()
	return snproxy.NewSnapshotter(snapshotsapi.NewSnapshotsClient(c.conn), snapshotterName)
}