func (c *Controller) Snapshot(name string, labels map[string]string) (string, error) {
	log := logrus.WithFields(logrus.Fields{"volume": c.Name, "snapshot": name})
	log.Info("Starting snapshot")
	if ne, err := iutil.NewNamespaceExecutor(util.GetInitiatorNS()); err != nil {
		log.WithError(err).Errorf("WARNING: continue to snapshot for %v, but cannot sync due to cannot get the namespace executor", name)
	} else {
		log.Info("Requesting system sync before snapshot")
		if _, err := ne.ExecuteWithTimeout(syncTimeout, "sync", []string{}); err != nil {
			log.WithError(err).Errorf("WARNING: failed to sync continuing with snapshot for %v", name)
		}
	}
	c.Lock()
	defer c.Unlock()
	if name == "" {
		name = util.UUID()
	}
	if remain, err := c.backend.RemainSnapshots(); err != nil {
		return "", err
	} else if remain <= 0 {
		return "", fmt.Errorf("too many snapshots created")
	}
	created := util.Now()
	if err := c.handleErrorNoLock(c.backend.Snapshot(name, true, created, labels)); err != nil {
		return "", err
	}
	log.Info("Finished snapshot")
	return name, nil
}
Snapshot will try to freeze the filesystem of the volume if possible
and will fallback to a system level sync in all other cases
package controller
import (
	"fmt"
	"net"
	"sync"
	"sync/atomic"
	"time"
	"github.com/pkg/errors"
	"github.com/sirupsen/logrus"
	"google.golang.org/grpc"
	iutil "github.com/longhorn/go-iscsi-helper/util"
	"github.com/longhorn/longhorn-engine/pkg/types"
	"github.com/longhorn/longhorn-engine/pkg/util"
)
type Controller struct {
	sync.RWMutex
	Name       string
	size       int64
	sectorSize int64
	replicas   []types.Replica
	factory    types.BackendFactory
	backend    *replicator
	frontend   types.Frontend
	isUpgrade  bool
	isExpanding             bool
	revisionCounterDisabled bool
	salvageRequested        bool
	GRPCAddress string
	GRPCServer  *grpc.Server
	ShutdownWG sync.WaitGroup
	lastError  error
	latestMetrics *types.Metrics
	metrics       *types.Metrics
	backupList      map[string]string
	backupListMutex *sync.RWMutex
	lastExpansionError    string
	lastExpansionFailedAt string
}
const (
	freezeTimeout         = 60 * time.Minute  
	syncTimeout           = 60 * time.Minute
	lastModifyCheckPeriod = 5 * time.Second
)
func NewController(name string, factory types.BackendFactory, frontend types.Frontend, isUpgrade bool, disableRevCounter bool, salvageRequested bool) *Controller {
	c := &Controller{
		factory:       factory,
		Name:          name,
		frontend:      frontend,
		metrics:       &types.Metrics{},
		latestMetrics: &types.Metrics{},
		backupList:      map[string]string{},
		backupListMutex: &sync.RWMutex{},
		isUpgrade:               isUpgrade,
		revisionCounterDisabled: disableRevCounter,
		salvageRequested:        salvageRequested,
	}
	c.reset()
	c.metricsStart()
	return c
}
func (c *Controller) StartGRPCServer() error {
	if c.GRPCServer == nil {
		return fmt.Errorf("cannot find grpc server")
	}
	grpcPort, err := util.GetPortFromAddress(c.GRPCAddress)
	if err != nil {
		return err
	}
	if c.GRPCAddress == "" || grpcPort == 0 {
		return fmt.Errorf("cannot find grpc address or port")
	}
	c.ShutdownWG.Add(1)
	go func() {
		defer c.ShutdownWG.Done()
		grpcAddress := c.GRPCAddress
		listener, err := net.Listen("tcp", c.GRPCAddress)
		if err != nil {
			logrus.Errorf("Failed to listen %v: %v", grpcAddress, err)
			c.lastError = err
			return
		}
		logrus.Infof("Listening on gRPC Controller server: %v", grpcAddress)
		err = c.GRPCServer.Serve(listener)
		logrus.Errorf("GRPC server at %v is down: %v", grpcAddress, err)
		c.lastError = err
	}()
	return nil
}
func (c *Controller) WaitForShutdown() error {
	c.ShutdownWG.Wait()
	return c.lastError
}
func (c *Controller) AddReplica(address string, snapshotRequired bool, mode types.Mode) error {
	return c.addReplica(address, snapshotRequired, mode)
}
func (c *Controller) hasWOReplica() bool {
	for _, i := range c.replicas {
		if i.Mode == types.WO {
			return true
		}
	}
	return false
}
func (c *Controller) canAdd(address string) (bool, error) {
	if c.hasReplica(address) {
		return false, nil
	}
	if c.hasWOReplica() {
		return false, fmt.Errorf("can only have one WO replica at a time")
	}
	if c.isExpanding {
		return false, fmt.Errorf("cannot add WO replica during expansion")
	}
	return true, nil
}
func (c *Controller) addReplica(address string, snapshotRequired bool, mode types.Mode) error {
	c.Lock()
	defer c.Unlock()
	if ok, err := c.canAdd(address); !ok {
		return err
	}
	newBackend, err := c.factory.Create(address)
	if err != nil {
		return err
	}
	replicaSize, err := newBackend.Size()
	if err != nil {
		return errors.Wrapf(err, "failed to get the size before adding a replica")
	}
	if c.size == 0 && len(c.replicas) == 0 {
		c.size = replicaSize
	}
	return c.addReplicaNoLock(newBackend, address, snapshotRequired, mode)
}