<commit_before>/*
Copyright 2017 The Rook Authors. All rights reserved.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

	http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package cmd

import (
	"encoding/json"
	"errors"
	"fmt"
	"net/rpc"
	"os"
	"strings"

	"github.com/rook/rook/pkg/daemon/ceph/agent/flexvolume"
	"github.com/spf13/cobra"
	k8smount "k8s.io/kubernetes/pkg/util/mount"
	"k8s.io/kubernetes/pkg/util/version"
	"k8s.io/kubernetes/pkg/volume/util"
)

const (
	mds_namespace_kernel_support = "4.7"
)

var (
	mountCmd = &cobra.Command{
		Use:   "mount",
		Short: "Mounts the volume to the pod volume",
		RunE:  handleMount,
	}
)

func init() {
	RootCmd.AddCommand(mountCmd)
}

func handleMount(cmd *cobra.Command, args []string) error {

	client, err := getRPCClient()
	if err != nil {
		return fmt.Errorf("Rook: Error getting RPC client: %v", err)
	}

	var opts = &flexvolume.AttachOptions{}
	if err := json.Unmarshal([]byte(args[1]), opts); err != nil {
		return fmt.Errorf("Rook: Could not parse options for mounting %s. Got %v", args[1], err)
	}
	opts.MountDir = args[0]

	if opts.FsType == cephFS {
		return mountCephFS(client, opts)
	}

	err = client.Call("Controller.GetAttachInfoFromMountDir", opts.MountDir, &opts)
	if err != nil {
		log(client, fmt.Sprintf("Attach volume %s/%s failed: %v", opts.Pool, opts.Image, err), true)
		return fmt.Errorf("Rook: Mount volume failed: %v", err)
	}

	// Attach volume to node
	devicePath, err := attach(client, opts)
	if err != nil {
		return err
	}

	// construct the input we'll need to get the global mount path
	driverDir, err := getDriverDir()
	if err != nil {
		return err
	}
	globalMountPathInput := flexvolume.GlobalMountPathInput{
		VolumeName: opts.VolumeName,
		DriverDir:  driverDir,
	}

	// Get global mount path
	var globalVolumeMountPath string
	err = client.Call("Controller.GetGlobalMountPath", globalMountPathInput, &globalVolumeMountPath)
	if err != nil {
		log(client, fmt.Sprintf("Attach volume %s/%s failed. Cannot get global volume mount path: %v", opts.Pool, opts.Image, err), true)
		return fmt.Errorf("Rook: Mount volume failed. Cannot get global volume mount path: %v", err)
	}

	mounter := getMounter()
	// Mount the volume to a global volume path
	err = mountDevice(client, mounter, devicePath, globalVolumeMountPath, opts)
	if err != nil {
		return err
	}

	// Mount the global mount path to pod mount dir
	err = mount(client, mounter, globalVolumeMountPath, opts)
	if err != nil {
		return err
	}
	log(client, fmt.Sprintf("volume %s/%s has been attached and mounted", opts.Pool, opts.Image), false)
	return nil
}

func attach(client *rpc.Client, opts *flexvolume.AttachOptions) (string, error) {

	log(client, fmt.Sprintf("calling agent to attach volume %s/%s", opts.Pool, opts.Image), false)
	var devicePath string
	err := client.Call("Controller.Attach", opts, &devicePath)
	if err != nil {
		log(client, fmt.Sprintf("Attach volume %s/%s failed: %v", opts.Pool, opts.Image, err), true)
		return "", fmt.Errorf("Rook: Mount volume failed: %v", err)
	}
	return devicePath, err
}

func mountDevice(client *rpc.Client, mounter *k8smount.SafeFormatAndMount, devicePath, globalVolumeMountPath string, opts *flexvolume.AttachOptions) error {
	notMnt, err := mounter.Interface.IsLikelyNotMountPoint(globalVolumeMountPath)
	if err != nil {
		if os.IsNotExist(err) {
			if err := os.MkdirAll(globalVolumeMountPath, 0750); err != nil {
				return fmt.Errorf("Rook: Mount volume failed. Cannot create global volume mount path dir: %v", err)
			}
			notMnt = true
		} else {
			return fmt.Errorf("Rook: Mount volume failed. Error checking if %s is a mount point: %v", globalVolumeMountPath, err)
		}
	}
	options := []string{opts.RW}
	if notMnt {
		err = redirectStdout(
			client,
			func() error {
				if err = mounter.FormatAndMount(devicePath, globalVolumeMountPath, opts.FsType, options); err != nil {
					return fmt.Errorf("failed to mount volume %s [%s] to %s, error %v", devicePath, opts.FsType, globalVolumeMountPath, err)
				}
				return nil
			},
		)
		if err != nil {<commit_msg>log the output of blkid to see if device should be formatted<commit_after>/*
Copyright 2017 The Rook Authors. All rights reserved.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

	http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package cmd

import (
	"encoding/json"
	"errors"
	"fmt"
	"net/rpc"
	"os"
	"strings"

	"github.com/rook/rook/pkg/daemon/ceph/agent/flexvolume"
	"github.com/rook/rook/pkg/util/exec"
	"github.com/spf13/cobra"
	k8smount "k8s.io/kubernetes/pkg/util/mount"
	"k8s.io/kubernetes/pkg/util/version"
	"k8s.io/kubernetes/pkg/volume/util"
)

const (
	mds_namespace_kernel_support = "4.7"
)

var (
	mountCmd = &cobra.Command{
		Use:   "mount",
		Short: "Mounts the volume to the pod volume",
		RunE:  handleMount,
	}
)

func init() {
	RootCmd.AddCommand(mountCmd)
}

func handleMount(cmd *cobra.Command, args []string) error {

	client, err := getRPCClient()
	if err != nil {
		return fmt.Errorf("Rook: Error getting RPC client: %v", err)
	}

	var opts = &flexvolume.AttachOptions{}
	if err := json.Unmarshal([]byte(args[1]), opts); err != nil {
		return fmt.Errorf("Rook: Could not parse options for mounting %s. Got %v", args[1], err)
	}
	opts.MountDir = args[0]

	if opts.FsType == cephFS {
		return mountCephFS(client, opts)
	}

	err = client.Call("Controller.GetAttachInfoFromMountDir", opts.MountDir, &opts)
	if err != nil {
		log(client, fmt.Sprintf("Attach volume %s/%s failed: %v", opts.Pool, opts.Image, err), true)
		return fmt.Errorf("Rook: Mount volume failed: %v", err)
	}

	// Attach volume to node
	devicePath, err := attach(client, opts)
	if err != nil {
		return err
	}

	// construct the input we'll need to get the global mount path
	driverDir, err := getDriverDir()
	if err != nil {
		return err
	}
	globalMountPathInput := flexvolume.GlobalMountPathInput{
		VolumeName: opts.VolumeName,
		DriverDir:  driverDir,
	}

	// Get global mount path
	var globalVolumeMountPath string
	err = client.Call("Controller.GetGlobalMountPath", globalMountPathInput, &globalVolumeMountPath)
	if err != nil {
		log(client, fmt.Sprintf("Attach volume %s/%s failed. Cannot get global volume mount path: %v", opts.Pool, opts.Image, err), true)
		return fmt.Errorf("Rook: Mount volume failed. Cannot get global volume mount path: %v", err)
	}

	mounter := getMounter()
	// Mount the volume to a global volume path
	err = mountDevice(client, mounter, devicePath, globalVolumeMountPath, opts)
	if err != nil {
		return err
	}

	// Mount the global mount path to pod mount dir
	err = mount(client, mounter, globalVolumeMountPath, opts)
	if err != nil {
		return err
	}
	log(client, fmt.Sprintf("volume %s/%s has been attached and mounted", opts.Pool, opts.Image), false)
	return nil
}

func attach(client *rpc.Client, opts *flexvolume.AttachOptions) (string, error) {

	log(client, fmt.Sprintf("calling agent to attach volume %s/%s", opts.Pool, opts.Image), false)
	var devicePath string
	err := client.Call("Controller.Attach", opts, &devicePath)
	if err != nil {
		log(client, fmt.Sprintf("Attach volume %s/%s failed: %v", opts.Pool, opts.Image, err), true)
		return "", fmt.Errorf("Rook: Mount volume failed: %v", err)
	}
	return devicePath, err
}

func mountDevice(client *rpc.Client, mounter *k8smount.SafeFormatAndMount, devicePath, globalVolumeMountPath string, opts *flexvolume.AttachOptions) error {
	notMnt, err := mounter.Interface.IsLikelyNotMountPoint(globalVolumeMountPath)
	if err != nil {
		if os.IsNotExist(err) {
			if err := os.MkdirAll(globalVolumeMountPath, 0750); err != nil {
				return fmt.Errorf("Rook: Mount volume failed. Cannot create global volume mount path dir: %v", err)
			}
			notMnt = true
		} else {
			return fmt.Errorf("Rook: Mount volume failed. Error checking if %s is a mount point: %v", globalVolumeMountPath, err)
		}
	}

	// Testing to see if the device is formatted. There has been observed an issue on slow systems where k8s believes there is no filesystem
	// formatted even though one does exist. K8s then proceeds to format the volume which would result in data loss. This logging is an attempt
	// to understand why k8s believes the volume is not formatted. See https://github.com/rook/rook/issues/1553
	log(client, fmt.Sprintf("Testing to see if device %s needs formatting...", devicePath), false)
	executor := &exec.CommandExecutor{}
	output, err := executor.ExecuteCommandWithOutput(false, "", "blkid", "-p", "-s", "TYPE", "-s", "PTTYPE", "-o", "export", devicePath)
	if err != nil {
		log(client, fmt.Sprintf("Formatting test (blkid -p -s TYPE -s PTTYPE -o export %s). Device may not be formatted. err: %+v", devicePath, err), false)
	} else {
		log(client, fmt.Sprintf("Formatting test (blkid -p -s TYPE -s PTTYPE -o export %s). Device is already formatted", devicePath), false)
	}
	if len(output) > 0 {
		log(client, fmt.Sprintf("Output: %s", output), false)
	}

	options := []string{opts.RW}
	if notMnt {
		err = redirectStdout(
			client,
			func() error {
				if err = mounter.FormatAndMount(devicePath, globalVolumeMountPath, opts.FsType, options); err != nil {
					return fmt.Errorf("failed to mount volume %s [%s] to %s, error %v", devicePath, opts.FsType, globalVolumeMountPath, err)
				}
				return nil
			},
		)
		if err != nil {
{'fraction_non_alphanumeric': 0.12258191867350968, 'fraction_numerical': 0.005527043031977891, 'mean_word_length': 4.183120204603581, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 4, 'https://': 1, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5492109', 'n_tokens_mistral': 3502, 'n_tokens_neox': 3303, 'n_words': 1101}