first commit

Author: bamboox

Dockerfile

@@ -0,0 +1,33 @@
+#FROM golang:1.12.4 as builder
+FROM registry.cn-hangzhou.aliyuncs.com/bamboo/golang:1.12.4 as builder
+
+ARG DIR=$GOPATH/src/github.com/kubernetes-sigs/san-client
+
+WORKDIR $DIR
+
+RUN mkdir -p $DIR
+
+COPY / $DIR
+
+ENV  GO111MODULE=on
+# setting go mod proxy, proxy run on onebox
+ENV  GOPROXY=https://goproxy.io
+#RUN  go mod tidy
+#RUN  go mod vendor
+#ENV  GO111MODULE=off
+
+RUN  go build -o main -ldflags '-s -w' -v $DIR/cmd/san-client-provisioner
+
+FROM alpine:3.9
+
+ARG APK_MIRROR=mirrors.aliyun.com
+RUN sed -i "s/dl-cdn.alpinelinux.org/${APK_MIRROR}/g" /etc/apk/repositories
+
+RUN apk add --no-cache libc6-compat
+## DON'T modify above, as it's common for all Alpine based parent and docker caching layer will used
+
+CMD ["./main"]
+
+COPY --from=builder /go/src/github.com/kubernetes-sigs/san-client/main .
+
+

README.md

@@ -0,0 +1,28 @@
+## san 挂载问题排查	
+1. FailedMount
+
+Warning  FailedMount             11s (x6 over 50s)  kubelet, dbm02           MountVolume.WaitForAttach failed for volume "pvc-53918fcd-c350-11e9-8c87-50af732f4b85" : no fc disk found
+
+
+ls /dev/disk/by-id/ | grep 845 #
+
+ssh [email protected] lshostvdiskmap | grep -i 845
+
+## delete by SCSI ID
+scsi_id=${scsi_id}
+wwid=${wwid}
+devices=lsscsi -i |grep MCS | grep ":${scsi_id}]" | grep -v $wwid | awk '{print $1}'
+for i in $devices; do device=${i:1:-1} ; echo 1 >/sys/class/scsi_device/$device/device/delete ; done
+
+
+## format
+挂载如果是非 readonly 需要注意format问题，数据会丢失
+
+# Annotation
+
+- ThinRateAnnotation    = "volume.beta.kubernetes.io/thin-rate"
+
+- MountOptionAnnotation = "volume.beta.kubernetes.io/mount-options"
+ 
+- MkfsFsTypeAnnotation  = "volume.beta.kubernetes.io/mkfs-fstype"  
+  default ext4

cmd/san-client-provisioner/provisioner.go

@@ -0,0 +1,278 @@
+package main
+
+import (
+	"flag"
+	"fmt"
+	log "github.com/golang/glog"
+	"github.com/kubernetes-sigs/san-client/fc"
+	"k8s.io/api/core/v1"
+	storage "k8s.io/api/storage/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/kubernetes/scheme"
+	corev1 "k8s.io/client-go/kubernetes/typed/core/v1"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/record"
+	volumehelpers "k8s.io/cloud-provider/volume/helpers"
+	"k8s.io/klog"
+	"k8s.io/kubernetes/pkg/apis/core/v1/helper"
+	"os"
+	"sigs.k8s.io/sig-storage-lib-external-provisioner/controller"
+	"strconv"
+	"strings"
+)
+
+const (
+	ThinRateAnnotation    = "volume.beta.kubernetes.io/mount-options"
+	MountOptionAnnotation = "volume.beta.kubernetes.io/mkfs-fstype"
+	MkfsFsTypeAnnotation  = "volume.beta.kubernetes.io/thin-rate"
+	DiskNameAnnotation    = "volume.beta.kubernetes.io/disk-name"
+
+	EventComponent = "san-client-provisioner"
+)
+
+type sanProvisioner struct {
+	client        kubernetes.Interface
+	eventRecorder record.EventRecorder
+}
+
+// NewSanProvisioner creates a new san provisioner
+func NewSanProvisioner(client kubernetes.Interface) controller.Provisioner {
+
+	//TODO: remove this
+	v1.AddToScheme(scheme.Scheme)
+	broadcaster := record.NewBroadcaster()
+	broadcaster.StartLogging(klog.Infof)
+	broadcaster.StartRecordingToSink(&corev1.EventSinkImpl{Interface: client.CoreV1().Events(v1.NamespaceAll)})
+	eventRecorder := broadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: EventComponent})
+
+	return &sanProvisioner{
+		client:        client,
+		eventRecorder: eventRecorder,
+	}
+}
+
+var _ controller.Provisioner = &sanProvisioner{}
+
+// Provision creates a storage asset and returns a PV object representing it.
+func (p *sanProvisioner) Provision(options controller.ProvisionOptions) (*v1.PersistentVolume, error) {
+	log.V(5).Infof("Provision options: %+v", options)
+	// step1. create disk
+	volSizeBytes := options.PVC.Spec.Resources.Requests[v1.ResourceName(v1.ResourceStorage)]
+	capacity, err := volumehelpers.RoundUpToGiBInt(volSizeBytes)
+	if err != nil {
+		return nil, err
+	}
+	params := options.StorageClass.Parameters
+	fc := fc.NewFcClient(params["user"], params["password"], params["sanserver"], 22, params["sanCliPipelineEndpoint"])
+	thinRate := ""
+	if _, ok := options.PVC.Annotations[ThinRateAnnotation]; ok {
+		thinRate = options.PVC.Annotations[ThinRateAnnotation]
+	}
+	volumeName := p.genVolumeName(options.PVName, options.PVC)
+	createErr := fc.CreateVolume(volumeName, strconv.Itoa(capacity), thinRate)
+	if createErr != nil {
+		return nil, createErr
+	}
+	// step2. GetWWIdByVolume
+	wwid, pvcErr := fc.GetWWIdByVolume(volumeName)
+	if pvcErr != nil {
+		return nil, pvcErr
+	}
+	log.V(5).Infof("GetWWIdByVolume wwid: %s", wwid)
+	// step3. mapHostForVolume
+	// TODO bind node
+	if options.SelectedNode != nil {
+		nodeName := options.SelectedNode.Name
+		if _, ok := options.SelectedNode.Labels["kubernetes.io/hostname"]; ok {
+			nodeName = options.SelectedNode.Labels["kubernetes.io/hostname"]
+		}
+		log.Infof("SelectedNode nodeName: %s", nodeName)
+	}
+
+	err = fc.MapAllHostForVolume(volumeName)
+	if err != nil {
+		return nil, err
+	}
+
+	// step4. removeAbnormalDevice
+
+	//get all k8s nodeAddress
+	nodes, err := p.client.CoreV1().Nodes().List(metav1.ListOptions{})
+	if err != nil {
+		log.Errorf("get nodes err :%v", err)
+		return nil, err
+	}
+	nodesIp, err := getAllNodeIP(nodes)
+	if err != nil || nodesIp == nil {
+		log.Errorf("getAllNodeIP nodesIp: %+v, err :%v", nodesIp, err)
+		return nil, err
+	}
+
+	err = fc.RemoveAbnormalDevice(nodesIp, volumeName, wwid)
+	if err != nil {
+		log.Errorf("RemoveAbnormalDevice err :%v", err)
+		return nil, err
+	}
+
+	// step5. create pv
+	// fsType default ext4 and  mountOptions
+	fsType := "ext4"
+	var mountOptions []string
+	if options.PVC.Annotations != nil {
+		if ft, ok := options.PVC.Annotations[MkfsFsTypeAnnotation]; ok {
+			fsType = ft
+		}
+		if mo, ok := options.PVC.Annotations[MountOptionAnnotation]; ok {
+			mountOptions = strings.Split(mo, ",")
+		}
+	}
+
+	pv := &v1.PersistentVolume{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: options.PVName,
+			Annotations: map[string]string{
+				DiskNameAnnotation: volumeName,
+			},
+		},
+		Spec: v1.PersistentVolumeSpec{
+			PersistentVolumeReclaimPolicy: *options.StorageClass.ReclaimPolicy,
+			AccessModes:                   options.PVC.Spec.AccessModes,
+			Capacity: v1.ResourceList{
+				v1.ResourceName(v1.ResourceStorage): options.PVC.Spec.Resources.Requests[v1.ResourceName(v1.ResourceStorage)],
+			},
+			PersistentVolumeSource: v1.PersistentVolumeSource{
+				FC: &v1.FCVolumeSource{
+					WWIDs:  []string{wwid},
+					FSType: fsType,
+				},
+			},
+			MountOptions: mountOptions,
+		},
+	}
+	// check disk is normal mapper
+	log.V(5).Infof("provision pv success")
+	return pv, nil
+}
+
+func (p *sanProvisioner) genVolumeName(pvName string, pvc *v1.PersistentVolumeClaim) string {
+	volumeName := fmt.Sprintf("%s_%s", pvc.Name, pvc.Namespace)
+	// FC SAN disk name max length 63
+	if len(volumeName) > 63 {
+		p.eventRecorder.Event(pvc, v1.EventTypeWarning, "genVolumeName", fmt.Sprintf("volumeName max length 63, this use pv name, vn: %s", pvName))
+		return pvName
+	}
+	return volumeName
+}
+
+func getAllNodeIP(nodes *v1.NodeList) (nodeIP map[string]string, err error) {
+	if len(nodes.Items) == 0 {
+		return nil, fmt.Errorf("nodes is nill")
+	}
+	nodeIP = make(map[string]string)
+	for _, node := range nodes.Items {
+		var name, ip string
+
+		name = node.Name
+		for _, addr := range node.Status.Addresses {
+			log.V(5).Infof("node: %s, addr: %+v", name, addr)
+			if addr.Type == v1.NodeInternalIP {
+				ip = addr.Address
+				break
+			}
+		}
+		if len(ip) == 0 {
+			return nil, fmt.Errorf("node: %s, not found NodeExternalIP", name)
+		}
+		// HACK: san host name
+		nodeIP[fmt.Sprintf("h_%s", name)] = ip
+	}
+	return
+}
+
+// Delete removes the storage asset that was created by Provision represented
+// by the given PV.
+func (p *sanProvisioner) Delete(volume *v1.PersistentVolume) error {
+	log.V(5).Infof("Delete volume: %+v", volume)
+
+	class, err := p.getClassForVolume(volume)
+	if err != nil {
+		return err
+	}
+	params := class.Parameters
+
+	volumeName := ""
+	if dn, ok := volume.Annotations[DiskNameAnnotation]; ok {
+		volumeName = dn
+	} else {
+		return fmt.Errorf("pv: %s, annotation not found disk name ", volume.Name)
+	}
+
+	fc := fc.NewFcClient(params["user"], params["password"], params["sanserver"], 22, params["sanCliPipelineEndpoint"])
+	log.V(5).Infof("DeleteVolume: begin delete san: %s, %s, %s, %s", volumeName, params["user"], params["password"], params["sanserver"])
+	// step1. RemoveAllMapping
+	err = fc.RemoveAllMapping(volumeName)
+	if err != nil {
+		return err
+	}
+	// step2. DeleteVolume
+	err = fc.DeleteVolume(volumeName)
+	if err != nil {
+		return err
+	}
+	log.V(5).Infof("provision delete pv success")
+	return nil
+}
+
+// getClassForVolume returns StorageClass
+func (p *sanProvisioner) getClassForVolume(pv *v1.PersistentVolume) (*storage.StorageClass, error) {
+
+	className := helper.GetPersistentVolumeClass(pv)
+	if className == "" {
+		return nil, fmt.Errorf("Volume has no storage class")
+	}
+	class, err := p.client.StorageV1().StorageClasses().Get(className, metav1.GetOptions{})
+	if err != nil {
+		return nil, err
+	}
+	return class, nil
+}
+
+func main() {
+	flag.Parse()
+	flag.Set("logtostderr", "true")
+	// Create an InClusterConfig and use it to create a client for the controller
+	// to use to communicate with Kubernetes
+	log.V(5).Infof("san provisioner start")
+	config, err := rest.InClusterConfig()
+	if err != nil {
+		log.Fatalf("Failed to create config: %v", err)
+	}
+
+	provisionerName := os.Getenv("PROVISIONER_NAME")
+	if provisionerName == "" {
+		log.Fatalf("environment variable %s is not set! Please set it.", provisionerName)
+	}
+
+	clientset, err := kubernetes.NewForConfig(config)
+	if err != nil {
+		log.Fatalf("Failed to create client: %v", err)
+	}
+
+	// The controller needs to know what the server version is because out-of-tree
+	// provisioners aren't officially supported until 1.5
+	serverVersion, err := clientset.Discovery().ServerVersion()
+	if err != nil {
+		log.Fatalf("Error getting server version: %v", err)
+	}
+
+	// Create the provisioner: it implements the Provisioner interface expected by
+	// the controller
+	sanProvisioner := NewSanProvisioner(clientset)
+
+	// Start the provision controller which will dynamically provision hostPath
+	pc := controller.NewProvisionController(clientset, provisionerName, sanProvisioner, serverVersion.GitVersion)
+	//
+	pc.Run(wait.NeverStop)
+}