// Validate ensures the structure of the policy is sane.
func (cfg *Policy) Validate() error {
	if cfg.FileSystems == nil {
		cfg.FileSystems = defaultFilesystems
	}

	if err := cfg.ValidateJSON(); err != nil {
		return errors.ErrJSONValidation.Combine(err)
	}

	if cfg.Backends == nil { // backend should be defined and its validated
		backends, ok := defaultDrivers[cfg.Backend]

		if !ok {
			return errored.Errorf("Invalid backend: %v", cfg.Backend)
		}
		cfg.Backends = backends
	}

	size, err := cfg.CreateOptions.ActualSize()
	if cfg.Backends.CRUD != "" && (size == 0 || err != nil) {
		return errored.Errorf("Size set to zero for non-empty CRUD backend %v", cfg.Backends.CRUD).Combine(err)
	}

	return nil
}

// CombineError is a simplication of errored.Combine
func CombineError(err error, format string, args ...interface{}) error {
	if erd, ok := err.(*errored.Error); ok {
		erd.Combine(errored.Errorf(format, args...))
	}

	return errored.New(err.Error()).Combine(errored.Errorf(format, args...))
}

func getConfig(c *cli.Context) (*manager.Config, string, error) {
	var reader io.Reader
	configFile := ""
	if !c.GlobalIsSet("config") {
		logrus.Debugf("no configuration was specified, starting with default.")
	} else if c.GlobalString("config") == "-" {
		logrus.Debugf("reading configuration from stdin")
		reader = bufio.NewReader(os.Stdin)
	} else {
		f, err := os.Open(c.GlobalString("config"))
		if err != nil {
			return nil, "", errored.Errorf("failed to open config file. Error: %v", err)
		}
		defer func() { f.Close() }()
		logrus.Debugf("reading configuration from file: %q", c.GlobalString("config"))
		reader = bufio.NewReader(f)
		configFile = c.GlobalString("config")
	}
	config := manager.DefaultConfig()
	if reader != nil {
		if _, err := config.MergeFromReader(reader); err != nil {
			return nil, "", errored.Errorf("failed to merge configuration. Error: %v", err)
		}
	}
	return config, configFile, nil
}

// Return major device number of the given kernel driver
func getDevID(kernelDriver string) (uint, error) {
	content, err := ioutil.ReadFile(deviceInfoFile)
	if err != nil {
		return 0, err
	}

	lines := strings.Split(string(content), "\n")
	blockDevs := false
	for _, line := range lines {
		if !blockDevs {
			blockDevs = line == "Block devices:"
			continue
		}

		if strings.HasSuffix(line, kernelDriver) {
			parts := strings.Split(line, " ")
			if len(parts) != 2 {
				return 0, errored.Errorf("Invalid input from file %q", deviceInfoFile)
			}

			majorID, err := convertToUint(parts[0])
			if err != nil {
				return 0, errored.Errorf("Invalid deviceID %q from device info file for kernel driver %q", parts, kernelDriver).Combine(err)
			}
			return majorID, nil
		}
	}

	return 0, errored.Errorf("Invalid kernel driver: %q", kernelDriver).Combine(errors.ErrDevNotFound)
}

// Mount records mounted volume
func (d *Driver) Mount(do storage.DriverOptions) (*storage.Mount, error) {
	d.logStat(getFunctionName())
	if err := os.MkdirAll(d.BaseMountPath, 0700); err != nil && !os.IsExist(err) {
		return nil, errored.Errorf("error creating %q directory: %v", d.BaseMountPath, err)
	}

	volumePath := filepath.Join(d.BaseMountPath, do.Volume.Params["pool"], do.Volume.Name)
	if err := os.MkdirAll(volumePath, 0700); err != nil && !os.IsExist(err) {
		return nil, errored.Errorf("error creating %q directory: %v", volumePath, err)
	}

	mount := &storage.Mount{
		Path:   volumePath,
		Volume: do.Volume,
	}

	content, err := json.Marshal(mount)
	if err != nil {
		return nil, err
	}

	_, err = d.client.Set(context.Background(), path.Join(mountedPrefix, do.Volume.Name), string(content), &client.SetOptions{PrevExist: client.PrevNoExist})
	logrus.Infof("%v %v", path.Join(mountedPrefix, do.Volume.Name), err)
	if err != nil {
		return nil, err
	}

	return mount, nil
}

// SetStatus updates the status and/or state of an asset in the inventory after
// performing lifecyslce related validations.
func (a *Asset) SetStatus(status AssetStatus, state AssetState) error {
	if a.status == status && a.state == state {
		logrus.Infof("asset already in status: %q and state: %q, no action required", status, state)
		return nil
	}

	if _, ok := lifecycleStatus[a.status][status]; !ok && a.status != status {
		return errored.Errorf("transition from %q to %q is not allowed", a.status, status)
	}

	if _, ok := lifecycleStates[status][state]; !ok {
		return errored.Errorf("%q is not a valid state when asset is in %q status", state, status)
	}

	if err := a.client.SetAssetStatus(a.name, status.String(), state.String(), StateDescription[state]); err != nil {
		return err
	}

	a.prevStatus = a.status
	a.prevState = a.state
	a.status = status
	a.state = state

	return nil
}

// Mounted shows any volumes that belong to volplugin on the host, in
// their native representation. They yield a *Mount.
func (d *Driver) Mounted(time.Duration) ([]*storage.Mount, error) {
	mounts := []*storage.Mount{}
	fis, err := ioutil.ReadDir(d.mountpath)
	if os.IsNotExist(err) {
		return mounts, os.MkdirAll(d.mountpath, 0700)
	} else if err != nil {
		return nil, errored.Errorf("Reading policy tree for mounts").Combine(err)
	}

	for _, fi := range fis {
		volumes, err := ioutil.ReadDir(filepath.Join(d.mountpath, fi.Name()))
		if err != nil {
			return nil, errored.Errorf("Reading mounted volumes for policy %q", fi.Name()).Combine(err)
		}

		for _, vol := range volumes {
			rel := filepath.Join(d.mountpath, fi.Name(), vol.Name())
			if err != nil {
				return nil, errored.Errorf("Calculating mount information for %q/%q", fi.Name(), vol.Name()).Combine(err)
			}

			mounts = append(mounts, &storage.Mount{
				Path: rel,
				Volume: storage.Volume{
					Name:   rel,
					Source: "null",
				},
			})
		}
	}

	return mounts, nil
}

func setValueWithType(field *reflect.Value, val string) error {
	if !field.CanSet() {
		return errored.Errorf("Cannot set value %q for struct element %q", val, field.Kind().String())
	}

	// navigate the kinds using the reflect types. fallthrough until we can get
	// at a convertible type. If nothing is applicable, error out.
	switch field.Kind() {
	case reflect.Int:
		fallthrough
	case reflect.Int32:
		return castInt32(field, val)
	case reflect.Int64:
		return castInt64(field, val)
	case reflect.Uint:
		fallthrough
	case reflect.Uint32:
		return castUint32(field, val)
	case reflect.Uint64:
		return castUint64(field, val)
	case reflect.Bool:
		return castBool(field, val)
	case reflect.Ptr:
		return castPtr(field, val)
	case reflect.String:
		return castString(field, val)
	}

	return errored.Errorf("Could not find appropriate type %q", field.Kind().String())
}

// CopySnapshot copies a snapshot into a new volume. Takes a DriverOptions,
// snap and volume name (string). Returns error on failure.
func (c *Driver) CopySnapshot(do storage.DriverOptions, snapName, newName string) error {
	intOrigName, err := c.internalName(do.Volume.Name)
	if err != nil {
		return err
	}

	intNewName, err := c.internalName(newName)
	if err != nil {
		return err
	}

	poolName := do.Volume.Params["pool"]

	list, err := c.List(storage.ListOptions{Params: storage.Params{"pool": poolName}})
	for _, vol := range list {
		if intNewName == vol.Name {
			return errored.Errorf("Volume %q already exists", vol.Name)
		}
	}

	errChan := make(chan error, 1)

	cmd := exec.Command("rbd", "snap", "protect", mkpool(poolName, intOrigName), "--snap", snapName)
	er, err := runWithTimeout(cmd, do.Timeout)

	// EBUSY indicates that the snapshot is already protected.
	if err != nil && er.ExitStatus != 0 && er.ExitStatus != int(unix.EBUSY) {
		if er.ExitStatus == int(unix.EEXIST) {
			err = errored.Errorf("Volume %q or snapshot name %q already exists. Snapshots cannot share the same name as the target volume.", do.Volume.Name, snapName).Combine(errors.Exists).Combine(errors.SnapshotProtect)
		}
		errChan <- err
		return err
	}

	defer c.cleanupCopy(snapName, newName, do, errChan)

	cmd = exec.Command("rbd", "clone", mkpool(poolName, intOrigName), mkpool(poolName, intNewName), "--snap", snapName)
	er, err = runWithTimeout(cmd, do.Timeout)
	if err != nil && er.ExitStatus == 0 {
		var err2 *errored.Error
		var ok bool

		err2, ok = err.(*errored.Error)
		if !ok {
			err2 = errored.New(err.Error())
		}
		errChan <- err2.Combine(errors.SnapshotCopy)
		return err2
	}

	if er.ExitStatus != 0 {
		newerr := errored.Errorf("Cloning snapshot to volume (volume %q, snapshot %q): %v", intOrigName, snapName, err).Combine(errors.SnapshotCopy).Combine(errors.SnapshotProtect)
		if er.ExitStatus != int(unix.EEXIST) {
			errChan <- newerr
		}
		return err
	}

	return nil
}

func (dc *DaemonConfig) getMounted() (map[string]*storage.Mount, map[string]int, error) {
	mounts := map[string]*storage.Mount{}
	counts := map[string]int{}

	now := time.Now()

	// XXX this loop will indefinitely run if the docker service is down.
	// This is intentional to ensure we don't take any action when docker is down.
	for {
		dockerClient, err := client.NewEnvClient()
		if err != nil {
			return nil, nil, errored.Errorf("Could not initiate docker client").Combine(err)
		}

		containers, err := dockerClient.ContainerList(context.Background(), types.ContainerListOptions{})
		if err != nil {
			if now.Sub(time.Now()) > dc.Global.Timeout {
				panic("Cannot contact docker")
			}
			logrus.Error(errored.Errorf("Could not query docker; retrying").Combine(err))
			time.Sleep(time.Second)
			continue
		}

		for _, container := range containers {
			if container.State == "running" {
				for _, mount := range container.Mounts {
					if mount.Driver == dc.PluginName {
						mounts[mount.Name] = nil
						counts[mount.Name]++
					}
				}
			}
		}

		break
	}

	for driverName := range backend.MountDrivers {
		cd, err := backend.NewMountDriver(driverName, dc.Global.MountPath)
		if err != nil {
			return nil, nil, err
		}

		mounted, err := cd.Mounted(dc.Global.Timeout)
		if err != nil {
			return nil, nil, err
		}

		for _, mount := range mounted {
			logrus.Debugf("Refreshing existing mount for %q: %v", mount.Volume.Name, *mount)
			mounts[mount.Volume.Name] = mount
		}
	}

	return mounts, counts, nil
}

func (d *DaemonConfig) handleCreate(w http.ResponseWriter, r *http.Request) {
	content, err := ioutil.ReadAll(r.Body)
	if err != nil {
		api.RESTHTTPError(w, errors.ReadBody.Combine(err))
		return
	}

	req := &config.VolumeRequest{}

	if err := json.Unmarshal(content, req); err != nil {
		api.RESTHTTPError(w, errors.UnmarshalRequest.Combine(err))
		return
	}

	if req.Policy == "" {
		api.RESTHTTPError(w, errors.GetPolicy.Combine(errored.Errorf("policy was blank")))
		return
	}

	if req.Name == "" {
		api.RESTHTTPError(w, errors.GetVolume.Combine(errored.Errorf("volume was blank")))
		return
	}

	hostname, err := os.Hostname()
	if err != nil {
		api.RESTHTTPError(w, errors.GetHostname.Combine(err))
		return
	}

	policy, err := d.Config.GetPolicy(req.Policy)
	if err != nil {
		api.RESTHTTPError(w, errors.GetPolicy.Combine(errored.New(req.Policy).Combine(err)))
		return
	}

	uc := &config.UseMount{
		Volume:   strings.Join([]string{req.Policy, req.Name}, "/"),
		Reason:   lock.ReasonCreate,
		Hostname: hostname,
	}

	snapUC := &config.UseSnapshot{
		Volume: strings.Join([]string{req.Policy, req.Name}, "/"),
		Reason: lock.ReasonCreate,
	}

	err = lock.NewDriver(d.Config).ExecuteWithMultiUseLock(
		[]config.UseLocker{uc, snapUC},
		d.Global.Timeout,
		d.createVolume(w, req, policy),
	)
	if err != nil && err != errors.Exists {
		api.RESTHTTPError(w, errors.CreateVolume.Combine(err))
		return
	}
}

// Validate validates volume options to ensure they are compatible with all
// storage drivers.
func (v Volume) Validate() error {
	if v.Name == "" {
		return errored.Errorf("Name is missing in storage driver")
	}

	if v.Params == nil {
		return errored.Errorf("Params are nil in storage driver")
	}

	return nil
}

// NewMountDriver instantiates and return a mount driver instance of the
// specified type
func NewMountDriver(backend, mountpath string) (storage.MountDriver, error) {
	f, ok := MountDrivers[backend]
	if !ok {
		return nil, errored.Errorf("invalid mount driver backend: %q", backend)
	}

	if mountpath == "" {
		return nil, errored.Errorf("mount path not specified, cannot continue")
	}

	return f(mountpath)
}

// Unmount a volume
func (d *Driver) Unmount(do storage.DriverOptions) error {
	mp, err := d.MountPath(do)
	if err != nil {
		return errored.Errorf("Calculating mount path for %q", do.Volume.Name).Combine(err)
	}

	if err := os.RemoveAll(mp); err != nil {
		return errored.Errorf("Removing mount path for %q", do.Volume.Name).Combine(err)
	}

	return nil
}

// Mount a volume. Returns the rbd device and mounted filesystem path.
// If you pass in the params what filesystem to use as `filesystem`, it will
// prefer that to `ext4` which is the default.
func (c *Driver) Mount(do storage.DriverOptions) (*storage.Mount, error) {
	intName, err := c.internalName(do.Volume.Name)
	if err != nil {
		return nil, err
	}

	poolName := do.Volume.Params["pool"]

	volumePath, err := c.mkMountPath(poolName, intName)
	if err != nil {
		return nil, err
	}

	devName, err := c.mapImage(do)
	if err != nil {
		return nil, err
	}

	// Create directory to mount
	if err := os.MkdirAll(c.mountpath, 0700); err != nil && !os.IsExist(err) {
		return nil, errored.Errorf("error creating %q directory: %v", c.mountpath, err)
	}

	if err := os.MkdirAll(volumePath, 0700); err != nil && !os.IsExist(err) {
		return nil, errored.Errorf("error creating %q directory: %v", volumePath, err)
	}

	// Obtain the major and minor node information about the device we're mounting.
	// This is critical for tuning cgroups and obtaining metrics for this device only.
	fi, err := os.Stat(devName)
	if err != nil {
		return nil, errored.Errorf("Failed to stat rbd device %q: %v", devName, err)
	}

	rdev := fi.Sys().(*syscall.Stat_t).Rdev

	major := rdev >> 8
	minor := rdev & 0xFF

	// Mount the RBD
	if err := unix.Mount(devName, volumePath, do.FSOptions.Type, 0, ""); err != nil {
		return nil, errored.Errorf("Failed to mount RBD dev %q: %v", devName, err)
	}

	return &storage.Mount{
		Device:   devName,
		Path:     volumePath,
		Volume:   do.Volume,
		DevMajor: uint(major),
		DevMinor: uint(minor),
	}, nil
}

func (m *Manager) reparseConfig() (*Config, error) {
	f, err := os.Open(m.configFile)
	if err != nil {
		return nil, errored.Errorf("failed to open config file. Error: %v", err)
	}
	defer func() { f.Close() }()
	logrus.Debugf("re-reading configuration from file: %q", m.configFile)
	reader := bufio.NewReader(f)
	config, err := DefaultConfig().MergeFromReader(reader)
	if err != nil {
		return nil, errored.Errorf("failed to merge configuration. Error: %v", err)
	}
	return config, nil
}

// CreateVolume creates a volume from parameters, including the policy to copy.
func CreateVolume(vr *VolumeRequest) (*Volume, error) {
	if vr.Name == "" {
		return nil, errored.Errorf("Volume name was empty").Combine(errors.InvalidVolume)
	}

	if vr.Policy == nil {
		return nil, errored.Errorf("Policy for volume %q was nil", vr.Name).Combine(errors.InvalidVolume)
	}

	var mount string

	if vr.Options != nil {
		mount = vr.Options["mount"]
		delete(vr.Options, "mount")
	}

	if err := merge.Opts(vr.Policy, vr.Options); err != nil {
		return nil, err
	}

	if vr.Policy.DriverOptions == nil {
		vr.Policy.DriverOptions = map[string]string{}
	}

	if err := vr.Policy.Validate(); err != nil {
		return nil, err
	}

	vc := &Volume{
		Backends:       vr.Policy.Backends,
		DriverOptions:  vr.Policy.DriverOptions,
		CreateOptions:  vr.Policy.CreateOptions,
		RuntimeOptions: vr.Policy.RuntimeOptions,
		Unlocked:       vr.Policy.Unlocked,
		PolicyName:     vr.Policy.Name,
		VolumeName:     vr.Name,
		MountSource:    mount,
	}

	if err := vc.Validate(); err != nil {
		return nil, err
	}

	if vc.CreateOptions.FileSystem == "" {
		vc.CreateOptions.FileSystem = DefaultFilesystem
	}

	return vc, nil
}

// Run executes the migration using a supplied backend.Backend
func (m *Migration) Run(b backend.Backend) error {
	fmt.Printf("Running migration #%d against %s backend\n", m.Version, b.Name())

	if err := m.runner(b); err != nil {
		return errored.Errorf("Encountered error during migration %d", m.Version).Combine(err)
	}

	if err := b.UpdateSchemaVersion(m.Version); err != nil {
		return errored.Errorf("Successfully applied migration but failed to update schema version key").Combine(err)
	}

	fmt.Println("")

	return nil
}

func (e *commissionEvent) eventValidate() error {
	var err error
	e._enodes, err = e.mgr.commonEventValidate(e.nodeNames)
	if err != nil {
		return err
	}

	if !IsValidHostGroup(e.hostGroup) {
		return errored.Errorf("invalid or empty host-group specified: %q", e.hostGroup)
	}

	// when workers are being configured, make sure that there is atleast one service-master
	if e.hostGroup == ansibleWorkerGroupName {
		masterCommissioned := false
		for name := range e.mgr.nodes {
			if _, ok := e._enodes[name]; ok {
				// skip nodes in the event
				continue
			}

			isDiscoveredAndAllocated, err := e.mgr.isDiscoveredAndAllocatedNode(name)
			if err != nil || !isDiscoveredAndAllocated {
				if err != nil {
					logrus.Debugf("a node check failed for %q. Error: %s", name, err)
				}
				// skip hosts that are not yet provisioned or not in discovered state
				continue
			}

			isMasterNode, err := e.mgr.isMasterNode(name)
			if err != nil || !isMasterNode {
				if err != nil {
					logrus.Debugf("a node check failed for %q. Error: %s", name, err)
				}
				//skip the hosts that are not in master group
				continue
			}

			// found a master node
			masterCommissioned = true
			break
		}
		if !masterCommissioned {
			return errored.Errorf("Cannot commission a worker node without existence of a master node in the cluster, make sure atleast one master node is commissioned.")
		}
	}
	return nil
}

// Path returns the path to the policy in the DB.
func (p *Policy) Path() (string, error) {
	if p.Name == "" {
		return "", errored.Errorf("Name is blank for this policy").Combine(errors.InvalidDBPath)
	}

	return strings.Join([]string{p.Prefix(), p.Name}, "/"), nil
}