From a986de98b0b3529c15ccef804367100929bb1f96 Mon Sep 17 00:00:00 2001 From: Josh Date: Sat, 6 Jun 2026 06:15:45 +0330 Subject: [PATCH] feat: init --- .gitignore | 2 + Dockerfile | 14 ++ charts/kloud-csi/Chart.yaml | 14 ++ charts/kloud-csi/templates/_helpers.tpl | 77 ++++++++ charts/kloud-csi/templates/configmap.yaml | 11 ++ .../templates/controller-deployment.yaml | 102 +++++++++++ charts/kloud-csi/templates/csidriver.yaml | 11 ++ .../kloud-csi/templates/node-daemonset.yaml | 130 ++++++++++++++ .../kloud-csi/templates/rbac-controller.yaml | 60 +++++++ charts/kloud-csi/templates/rbac-node.yaml | 36 ++++ charts/kloud-csi/templates/secret.yaml | 13 ++ charts/kloud-csi/templates/storageclass.yaml | 16 ++ charts/kloud-csi/values.yaml | 76 ++++++++ config/config.go | 140 +++++++++++++++ config/config_test.go | 94 ++++++++++ config/env.go | 76 ++++++++ examples/csi-client.hcl | 14 ++ go.mod | 38 ++++ go.sum | 73 ++++++++ main.go | 66 +++++++ pkg/csi/api/client.go | 166 +++++++++++++++++ pkg/csi/api/client_test.go | 62 +++++++ pkg/csi/client/backend.go | 15 ++ pkg/csi/driver/controller.go | 168 ++++++++++++++++++ pkg/csi/driver/driver.go | 149 ++++++++++++++++ pkg/csi/driver/identity.go | 82 +++++++++ pkg/csi/driver/mount_linux.go | 120 +++++++++++++ pkg/csi/driver/mount_other.go | 39 ++++ pkg/csi/driver/node.go | 165 +++++++++++++++++ pkg/csi/driver/volume.go | 7 + pkg/csi/provisioner/types.go | 51 ++++++ pkg/csi/volume/id.go | 29 +++ pkg/csi/volume/id_test.go | 25 +++ 33 files changed, 2141 insertions(+) create mode 100644 .gitignore create mode 100644 Dockerfile create mode 100644 charts/kloud-csi/Chart.yaml create mode 100644 charts/kloud-csi/templates/_helpers.tpl create mode 100644 charts/kloud-csi/templates/configmap.yaml create mode 100644 charts/kloud-csi/templates/controller-deployment.yaml create mode 100644 charts/kloud-csi/templates/csidriver.yaml create mode 100644 charts/kloud-csi/templates/node-daemonset.yaml create mode 100644 charts/kloud-csi/templates/rbac-controller.yaml create mode 100644 charts/kloud-csi/templates/rbac-node.yaml create mode 100644 charts/kloud-csi/templates/secret.yaml create mode 100644 charts/kloud-csi/templates/storageclass.yaml create mode 100644 charts/kloud-csi/values.yaml create mode 100644 config/config.go create mode 100644 config/config_test.go create mode 100644 config/env.go create mode 100644 examples/csi-client.hcl create mode 100644 go.mod create mode 100644 go.sum create mode 100644 main.go create mode 100644 pkg/csi/api/client.go create mode 100644 pkg/csi/api/client_test.go create mode 100644 pkg/csi/client/backend.go create mode 100644 pkg/csi/driver/controller.go create mode 100644 pkg/csi/driver/driver.go create mode 100644 pkg/csi/driver/identity.go create mode 100644 pkg/csi/driver/mount_linux.go create mode 100644 pkg/csi/driver/mount_other.go create mode 100644 pkg/csi/driver/node.go create mode 100644 pkg/csi/driver/volume.go create mode 100644 pkg/csi/provisioner/types.go create mode 100644 pkg/csi/volume/id.go create mode 100644 pkg/csi/volume/id_test.go diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..2dc7733 --- /dev/null +++ b/.gitignore @@ -0,0 +1,2 @@ +csi.hcl +*.hcl.local diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..407cb23 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,14 @@ +FROM golang:1.24-alpine AS build + +WORKDIR /build +RUN apk add --no-cache git +COPY go.mod go.sum ./ +RUN go mod download +COPY . . +RUN CGO_ENABLED=0 go build -o /kks-csi . + +FROM alpine:3.21 +RUN apk add --no-cache mount util-linux e2fsprogs findmnt +COPY --from=build /kks-csi /kks-csi +USER 0:0 +ENTRYPOINT ["/kks-csi"] diff --git a/charts/kloud-csi/Chart.yaml b/charts/kloud-csi/Chart.yaml new file mode 100644 index 0000000..dc6fe5d --- /dev/null +++ b/charts/kloud-csi/Chart.yaml @@ -0,0 +1,14 @@ +apiVersion: v2 +name: kloud-csi +description: Kloud CSI driver for kks persistent volumes +type: application +version: 0.1.0 +appVersion: "0.1.0" +kubeVersion: ">=1.28.0-0" +home: https://github.com/KubelanCloud/kks-csi-plugin +keywords: + - csi + - storage + - kloud +maintainers: + - name: Kloud Team diff --git a/charts/kloud-csi/templates/_helpers.tpl b/charts/kloud-csi/templates/_helpers.tpl new file mode 100644 index 0000000..134e244 --- /dev/null +++ b/charts/kloud-csi/templates/_helpers.tpl @@ -0,0 +1,77 @@ +{{/* +Expand the name of the chart. +*/}} +{{- define "kloud-csi.name" -}} +{{- default .Chart.Name .Values.nameOverride | trunc 63 | trimSuffix "-" }} +{{- end }} + +{{/* +Create a default fully qualified app name. +*/}} +{{- define "kloud-csi.fullname" -}} +{{- if .Values.fullnameOverride }} +{{- .Values.fullnameOverride | trunc 63 | trimSuffix "-" }} +{{- else }} +{{- $name := default .Chart.Name .Values.nameOverride }} +{{- if contains $name .Release.Name }} +{{- .Release.Name | trunc 63 | trimSuffix "-" }} +{{- else }} +{{- printf "%s-%s" .Release.Name $name | trunc 63 | trimSuffix "-" }} +{{- end }} +{{- end }} +{{- end }} + +{{- define "kloud-csi.chart" -}} +{{- printf "%s-%s" .Chart.Name .Chart.Version | replace "+" "_" | trunc 63 | trimSuffix "-" }} +{{- end }} + +{{- define "kloud-csi.labels" -}} +helm.sh/chart: {{ include "kloud-csi.chart" . }} +{{ include "kloud-csi.selectorLabels" . }} +{{- if .Chart.AppVersion }} +app.kubernetes.io/version: {{ .Chart.AppVersion | quote }} +{{- end }} +app.kubernetes.io/managed-by: {{ .Release.Service }} +{{- end }} + +{{- define "kloud-csi.selectorLabels" -}} +app.kubernetes.io/name: {{ include "kloud-csi.name" . }} +app.kubernetes.io/instance: {{ .Release.Name }} +{{- end }} + +{{- define "kloud-csi.controllerName" -}} +{{- printf "%s-controller" (include "kloud-csi.fullname" .) }} +{{- end }} + +{{- define "kloud-csi.nodeName" -}} +{{- printf "%s-node" (include "kloud-csi.fullname" .) }} +{{- end }} + +{{- define "kloud-csi.controllerServiceAccountName" -}} +{{- if .Values.serviceAccount.controller.create }} +{{- default (printf "%s-controller-sa" (include "kloud-csi.fullname" .)) .Values.serviceAccount.controller.name }} +{{- else }} +{{- default "default" .Values.serviceAccount.controller.name }} +{{- end }} +{{- end }} + +{{- define "kloud-csi.nodeServiceAccountName" -}} +{{- if .Values.serviceAccount.node.create }} +{{- default (printf "%s-node-sa" (include "kloud-csi.fullname" .)) .Values.serviceAccount.node.name }} +{{- else }} +{{- default "default" .Values.serviceAccount.node.name }} +{{- end }} +{{- end }} + +{{- define "kloud-csi.driverImage" -}} +{{- printf "%s:%s" .Values.image.repository (default .Chart.AppVersion .Values.image.tag) }} +{{- end }} + +{{- define "kloud-csi.validateRequired" -}} +{{- if and (not .Values.existingSecret) (not .Values.accessToken) }} +{{- fail "accessToken or existingSecret is required" }} +{{- end }} +{{- if not .Values.serverURL }} +{{- fail "serverURL is required" }} +{{- end }} +{{- end }} diff --git a/charts/kloud-csi/templates/configmap.yaml b/charts/kloud-csi/templates/configmap.yaml new file mode 100644 index 0000000..0d74f80 --- /dev/null +++ b/charts/kloud-csi/templates/configmap.yaml @@ -0,0 +1,11 @@ +apiVersion: v1 +kind: ConfigMap +metadata: + name: {{ include "kloud-csi.fullname" . }}-driver + namespace: {{ .Release.Namespace }} + labels: + {{- include "kloud-csi.labels" . | nindent 4 }} +data: + driver.hcl: | + driver {} + client {} diff --git a/charts/kloud-csi/templates/controller-deployment.yaml b/charts/kloud-csi/templates/controller-deployment.yaml new file mode 100644 index 0000000..3269881 --- /dev/null +++ b/charts/kloud-csi/templates/controller-deployment.yaml @@ -0,0 +1,102 @@ +{{- include "kloud-csi.validateRequired" . }} +apiVersion: apps/v1 +kind: Deployment +metadata: + name: {{ include "kloud-csi.controllerName" . }} + namespace: {{ .Release.Namespace }} + labels: + {{- include "kloud-csi.labels" . | nindent 4 }} + app.kubernetes.io/component: controller +spec: + replicas: {{ .Values.controller.replicas }} + selector: + matchLabels: + {{- include "kloud-csi.selectorLabels" . | nindent 6 }} + app.kubernetes.io/component: controller + template: + metadata: + labels: + {{- include "kloud-csi.selectorLabels" . | nindent 8 }} + app.kubernetes.io/component: controller + {{- with .Values.podLabels }} + {{- toYaml . | nindent 8 }} + {{- end }} + {{- with .Values.podAnnotations }} + annotations: + {{- toYaml . | nindent 8 }} + {{- end }} + spec: + serviceAccountName: {{ include "kloud-csi.controllerServiceAccountName" . }} + {{- with .Values.nodeSelector }} + nodeSelector: + {{- toYaml . | nindent 8 }} + {{- end }} + {{- with .Values.affinity }} + affinity: + {{- toYaml . | nindent 8 }} + {{- end }} + containers: + - name: kloud-csi-plugin + image: {{ include "kloud-csi.driverImage" . }} + imagePullPolicy: {{ .Values.image.pullPolicy }} + args: + - --config-file + - /etc/kloud-csi/driver.hcl + env: + - name: KKS_CSI_SERVER_URL + value: {{ .Values.serverURL | quote }} + - name: KKS_CSI_ACCESS_TOKEN + valueFrom: + secretKeyRef: + name: {{ default (printf "%s-config" (include "kloud-csi.fullname" .)) .Values.existingSecret }} + key: {{ .Values.existingSecretAccessTokenKey }} + - name: KKS_CSI_DRIVER_MODE + value: controller + - name: KKS_CSI_DRIVER_ENDPOINT + value: unix:///csi/csi.sock + - name: KKS_CSI_DRIVER_NAME + value: {{ .Values.driver.name | quote }} + volumeMounts: + - name: socket-dir + mountPath: /csi + - name: config + mountPath: /etc/kloud-csi + {{- with .Values.controller.resources }} + resources: + {{- toYaml . | nindent 12 }} + {{- end }} + - name: csi-provisioner + image: {{ .Values.sidecars.provisioner.repository }}:{{ .Values.sidecars.provisioner.tag }} + args: + - --csi-address=/csi/csi.sock + - --v=2 + - --feature-gates=Topology=false + - --timeout=60s + - --leader-election + - --default-fstype=ext4 + volumeMounts: + - name: socket-dir + mountPath: /csi + - name: csi-attacher + image: {{ .Values.sidecars.attacher.repository }}:{{ .Values.sidecars.attacher.tag }} + args: + - --csi-address=/csi/csi.sock + - --v=2 + - --leader-election + volumeMounts: + - name: socket-dir + mountPath: /csi + - name: liveness-probe + image: {{ .Values.sidecars.livenessProbe.repository }}:{{ .Values.sidecars.livenessProbe.tag }} + args: + - --csi-address=/csi/csi.sock + - --health-port=9808 + volumeMounts: + - name: socket-dir + mountPath: /csi + volumes: + - name: socket-dir + emptyDir: {} + - name: config + configMap: + name: {{ include "kloud-csi.fullname" . }}-driver diff --git a/charts/kloud-csi/templates/csidriver.yaml b/charts/kloud-csi/templates/csidriver.yaml new file mode 100644 index 0000000..be7dd69 --- /dev/null +++ b/charts/kloud-csi/templates/csidriver.yaml @@ -0,0 +1,11 @@ +apiVersion: storage.k8s.io/v1 +kind: CSIDriver +metadata: + name: {{ .Values.driver.name }} + labels: + {{- include "kloud-csi.labels" . | nindent 4 }} +spec: + attachRequired: true + podInfoOnMount: false + volumeLifecycleModes: + - Persistent diff --git a/charts/kloud-csi/templates/node-daemonset.yaml b/charts/kloud-csi/templates/node-daemonset.yaml new file mode 100644 index 0000000..defe5e2 --- /dev/null +++ b/charts/kloud-csi/templates/node-daemonset.yaml @@ -0,0 +1,130 @@ +{{- include "kloud-csi.validateRequired" . }} +apiVersion: apps/v1 +kind: DaemonSet +metadata: + name: {{ include "kloud-csi.nodeName" . }} + namespace: {{ .Release.Namespace }} + labels: + {{- include "kloud-csi.labels" . | nindent 4 }} + app.kubernetes.io/component: node +spec: + selector: + matchLabels: + {{- include "kloud-csi.selectorLabels" . | nindent 6 }} + app.kubernetes.io/component: node + template: + metadata: + labels: + {{- include "kloud-csi.selectorLabels" . | nindent 8 }} + app.kubernetes.io/component: node + {{- with .Values.podLabels }} + {{- toYaml . | nindent 8 }} + {{- end }} + {{- with .Values.podAnnotations }} + annotations: + {{- toYaml . | nindent 8 }} + {{- end }} + spec: + serviceAccountName: {{ include "kloud-csi.nodeServiceAccountName" . }} + priorityClassName: {{ .Values.node.priorityClassName }} + hostNetwork: true + hostPID: true + {{- with .Values.node.tolerations }} + tolerations: + {{- toYaml . | nindent 8 }} + {{- end }} + {{- with .Values.nodeSelector }} + nodeSelector: + {{- toYaml . | nindent 8 }} + {{- end }} + {{- with .Values.affinity }} + affinity: + {{- toYaml . | nindent 8 }} + {{- end }} + containers: + - name: kloud-csi-plugin + image: {{ include "kloud-csi.driverImage" . }} + imagePullPolicy: {{ .Values.image.pullPolicy }} + securityContext: + privileged: true + args: + - --config-file + - /etc/kloud-csi/driver.hcl + env: + - name: KKS_CSI_SERVER_URL + value: {{ .Values.serverURL | quote }} + - name: KKS_CSI_ACCESS_TOKEN + valueFrom: + secretKeyRef: + name: {{ default (printf "%s-config" (include "kloud-csi.fullname" .)) .Values.existingSecret }} + key: {{ .Values.existingSecretAccessTokenKey }} + - name: KKS_CSI_DRIVER_MODE + value: node + - name: KKS_CSI_DRIVER_ENDPOINT + value: unix:///csi/csi.sock + - name: KKS_CSI_DRIVER_NAME + value: {{ .Values.driver.name | quote }} + - name: KKS_CSI_NODE_ID + valueFrom: + fieldRef: + fieldPath: spec.nodeName + volumeMounts: + - name: plugin-dir + mountPath: /csi + - name: kubelet-dir + mountPath: /var/lib/kubelet + mountPropagation: Bidirectional + - name: device-dir + mountPath: /dev + - name: config + mountPath: /etc/kloud-csi + {{- with .Values.node.resources }} + resources: + {{- toYaml . | nindent 12 }} + {{- end }} + - name: node-driver-registrar + image: {{ .Values.sidecars.registrar.repository }}:{{ .Values.sidecars.registrar.tag }} + args: + - --v=2 + - --csi-address=/csi/csi.sock + - --kubelet-registration-path=/var/lib/kubelet/plugins/{{ .Values.driver.name }}/csi.sock + env: + - name: KUBE_NODE_NAME + valueFrom: + fieldRef: + fieldPath: spec.nodeName + volumeMounts: + - name: plugin-dir + mountPath: /csi + - name: registration-dir + mountPath: /registration + - name: kubelet-dir + mountPath: /var/lib/kubelet + mountPropagation: HostToContainer + - name: liveness-probe + image: {{ .Values.sidecars.livenessProbe.repository }}:{{ .Values.sidecars.livenessProbe.tag }} + args: + - --csi-address=/csi/csi.sock + - --health-port=9809 + volumeMounts: + - name: plugin-dir + mountPath: /csi + volumes: + - name: plugin-dir + hostPath: + path: /var/lib/kubelet/plugins/{{ .Values.driver.name }}/ + type: DirectoryOrCreate + - name: registration-dir + hostPath: + path: /var/lib/kubelet/plugins_registry/ + type: Directory + - name: kubelet-dir + hostPath: + path: /var/lib/kubelet + type: Directory + - name: device-dir + hostPath: + path: /dev + - name: config + configMap: + name: {{ include "kloud-csi.fullname" . }}-driver diff --git a/charts/kloud-csi/templates/rbac-controller.yaml b/charts/kloud-csi/templates/rbac-controller.yaml new file mode 100644 index 0000000..5aecc72 --- /dev/null +++ b/charts/kloud-csi/templates/rbac-controller.yaml @@ -0,0 +1,60 @@ +{{- if .Values.rbac.create }} +apiVersion: v1 +kind: ServiceAccount +metadata: + name: {{ include "kloud-csi.controllerServiceAccountName" . }} + namespace: {{ .Release.Namespace }} + labels: + {{- include "kloud-csi.labels" . | nindent 4 }} + app.kubernetes.io/component: controller +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: ClusterRole +metadata: + name: {{ include "kloud-csi.fullname" . }}-controller + labels: + {{- include "kloud-csi.labels" . | nindent 4 }} +rules: + - apiGroups: [""] + resources: ["persistentvolumes"] + verbs: ["get", "list", "watch", "create", "delete", "patch"] + - apiGroups: [""] + resources: ["persistentvolumeclaims"] + verbs: ["get", "list", "watch", "update", "patch"] + - apiGroups: [""] + resources: ["persistentvolumeclaims/status"] + verbs: ["update", "patch"] + - apiGroups: [""] + resources: ["nodes"] + verbs: ["get", "list", "watch"] + - apiGroups: ["storage.k8s.io"] + resources: ["storageclasses"] + verbs: ["get", "list", "watch"] + - apiGroups: [""] + resources: ["events"] + verbs: ["list", "watch", "create", "update", "patch"] + - apiGroups: ["storage.k8s.io"] + resources: ["volumeattachments"] + verbs: ["get", "list", "watch", "update", "patch"] + - apiGroups: ["storage.k8s.io"] + resources: ["volumeattachments/status"] + verbs: ["patch"] + - apiGroups: ["coordination.k8s.io"] + resources: ["leases"] + verbs: ["get", "watch", "list", "create", "update", "patch"] +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: ClusterRoleBinding +metadata: + name: {{ include "kloud-csi.fullname" . }}-controller + labels: + {{- include "kloud-csi.labels" . | nindent 4 }} +roleRef: + apiGroup: rbac.authorization.k8s.io + kind: ClusterRole + name: {{ include "kloud-csi.fullname" . }}-controller +subjects: + - kind: ServiceAccount + name: {{ include "kloud-csi.controllerServiceAccountName" . }} + namespace: {{ .Release.Namespace }} +{{- end }} diff --git a/charts/kloud-csi/templates/rbac-node.yaml b/charts/kloud-csi/templates/rbac-node.yaml new file mode 100644 index 0000000..f31fa15 --- /dev/null +++ b/charts/kloud-csi/templates/rbac-node.yaml @@ -0,0 +1,36 @@ +{{- if .Values.rbac.create }} +apiVersion: v1 +kind: ServiceAccount +metadata: + name: {{ include "kloud-csi.nodeServiceAccountName" . }} + namespace: {{ .Release.Namespace }} + labels: + {{- include "kloud-csi.labels" . | nindent 4 }} + app.kubernetes.io/component: node +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: ClusterRole +metadata: + name: {{ include "kloud-csi.fullname" . }}-node + labels: + {{- include "kloud-csi.labels" . | nindent 4 }} +rules: + - apiGroups: [""] + resources: ["nodes"] + verbs: ["get", "list", "watch"] +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: ClusterRoleBinding +metadata: + name: {{ include "kloud-csi.fullname" . }}-node + labels: + {{- include "kloud-csi.labels" . | nindent 4 }} +roleRef: + apiGroup: rbac.authorization.k8s.io + kind: ClusterRole + name: {{ include "kloud-csi.fullname" . }}-node +subjects: + - kind: ServiceAccount + name: {{ include "kloud-csi.nodeServiceAccountName" . }} + namespace: {{ .Release.Namespace }} +{{- end }} diff --git a/charts/kloud-csi/templates/secret.yaml b/charts/kloud-csi/templates/secret.yaml new file mode 100644 index 0000000..b0b59ca --- /dev/null +++ b/charts/kloud-csi/templates/secret.yaml @@ -0,0 +1,13 @@ +{{- include "kloud-csi.validateRequired" . }} +{{- if not .Values.existingSecret }} +apiVersion: v1 +kind: Secret +metadata: + name: {{ include "kloud-csi.fullname" . }}-config + namespace: {{ .Release.Namespace }} + labels: + {{- include "kloud-csi.labels" . | nindent 4 }} +type: Opaque +stringData: + access-token: {{ required "accessToken is required when existingSecret is not set" .Values.accessToken | quote }} +{{- end }} diff --git a/charts/kloud-csi/templates/storageclass.yaml b/charts/kloud-csi/templates/storageclass.yaml new file mode 100644 index 0000000..896297a --- /dev/null +++ b/charts/kloud-csi/templates/storageclass.yaml @@ -0,0 +1,16 @@ +{{- if .Values.storageClass.enabled }} +apiVersion: storage.k8s.io/v1 +kind: StorageClass +metadata: + name: {{ .Values.storageClass.name }} + {{- if .Values.storageClass.isDefault }} + annotations: + storageclass.kubernetes.io/is-default-class: "true" + {{- end }} + labels: + {{- include "kloud-csi.labels" . | nindent 4 }} +provisioner: {{ .Values.driver.name }} +reclaimPolicy: {{ .Values.storageClass.reclaimPolicy }} +volumeBindingMode: {{ .Values.storageClass.volumeBindingMode }} +allowVolumeExpansion: {{ .Values.storageClass.allowVolumeExpansion }} +{{- end }} diff --git a/charts/kloud-csi/values.yaml b/charts/kloud-csi/values.yaml new file mode 100644 index 0000000..a6e3efe --- /dev/null +++ b/charts/kloud-csi/values.yaml @@ -0,0 +1,76 @@ +# Kloud CSI driver — install on each user Kubernetes cluster. +# +# Required: +# serverURL — URL of the management CSI server (from your Kloud cluster details) +# accessToken — cluster csi_access_token (create a Kubernetes secret or set here) +# +# Example: +# helm install kloud-csi ./charts/kloud-csi \ +# --namespace kube-system \ +# --set serverURL=http://10.0.0.5:9766 \ +# --set accessToken="$(kubectl get secret ... -o jsonpath='{.data.token}' | base64 -d)" + +nameOverride: "" +fullnameOverride: "" + +image: + repository: ghcr.io/KubelanCloud/kks-csi-plugin + tag: "latest" + pullPolicy: IfNotPresent + +serverURL: "" +accessToken: "" +existingSecret: "" +existingSecretAccessTokenKey: access-token + +driver: + name: storage.csi.kloud.team + +sidecars: + provisioner: + repository: registry.k8s.io/sig-storage/csi-provisioner + tag: v5.1.0 + attacher: + repository: registry.k8s.io/sig-storage/csi-attacher + tag: v4.7.0 + registrar: + repository: registry.k8s.io/sig-storage/csi-node-driver-registrar + tag: v2.12.0 + livenessProbe: + repository: registry.k8s.io/sig-storage/livenessprobe + tag: v2.13.0 + +storageClass: + enabled: true + name: kloud-csi + isDefault: false + reclaimPolicy: Delete + volumeBindingMode: WaitForFirstConsumer + allowVolumeExpansion: false + +controller: + replicas: 1 + resources: {} + +node: + resources: {} + priorityClassName: system-node-critical + tolerations: + - operator: Exists + +serviceAccount: + controller: + create: true + name: "" + node: + create: true + name: "" + +rbac: + create: true + +podLabels: {} +podAnnotations: {} + +nodeSelector: {} +affinity: {} diff --git a/config/config.go b/config/config.go new file mode 100644 index 0000000..63be058 --- /dev/null +++ b/config/config.go @@ -0,0 +1,140 @@ +package config + +import ( + "fmt" + "os" + "strings" + "time" + + hcl "github.com/hashicorp/hcl/v2/hclsimple" +) + +const ( + CSIModeAll = "all" + CSIModeController = "controller" + CSIModeNode = "node" +) + +type Config struct { + Driver *DriverConf `hcl:"driver,block"` + Client *ClientConf `hcl:"client,block"` +} + +type DriverConf struct { + Name string `hcl:"name,optional"` + Endpoint string `hcl:"endpoint,optional"` + NodeID string `hcl:"node_id,optional"` + Mode string `hcl:"mode,optional"` +} + +type ClientConf struct { + ServerURL string `hcl:"server_url,optional"` + AccessToken string `hcl:"access_token,optional"` + Token string `hcl:"token,optional"` + TimeoutSeconds int `hcl:"timeout_seconds,optional"` +} + +func Load(filename string) (*Config, error) { + cfg := &Config{} + if err := hcl.DecodeFile(filename, nil, cfg); err != nil { + return nil, err + } + if err := cfg.normalize(); err != nil { + return nil, err + } + return cfg, nil +} + +// LoadClient loads driver config from file and/or env vars used by the Helm chart. +func LoadClient(filename string) (*Config, error) { + var cfg *Config + if filename != "" { + if _, err := os.Stat(filename); err == nil { + loaded, err := Load(filename) + if err != nil { + return nil, err + } + cfg = loaded + } + } + if cfg == nil { + cfg = &Config{ + Driver: &DriverConf{}, + Client: &ClientConf{}, + } + if err := cfg.normalize(); err != nil { + return nil, err + } + } + ApplyEnvOverrides(cfg) + ApplyClientTimeoutFromEnv(cfg) + if err := cfg.Validate(); err != nil { + return nil, err + } + return cfg, nil +} + +func (c *Config) Validate() error { + if c.Client == nil || strings.TrimSpace(c.Client.ServerURL) == "" { + return fmt.Errorf("client.server_url is required") + } + if strings.TrimSpace(c.Client.BearerToken()) == "" { + return fmt.Errorf("client.access_token is required") + } + return nil +} + +func (c *Config) normalize() error { + if c.Driver == nil { + c.Driver = &DriverConf{} + } + if c.Client == nil { + c.Client = &ClientConf{} + } + + if c.Driver.Name == "" { + c.Driver.Name = "storage.csi.kloud.team" + } + if c.Driver.Endpoint == "" { + c.Driver.Endpoint = "unix:///var/lib/kubelet/plugins/storage.csi.kloud.team/csi.sock" + } + if c.Driver.NodeID == "" { + hostname, err := os.Hostname() + if err != nil { + return fmt.Errorf("driver.node_id is empty and hostname lookup failed: %w", err) + } + c.Driver.NodeID = hostname + } + if c.Driver.Mode == "" { + c.Driver.Mode = CSIModeAll + } + + mode := strings.ToLower(strings.TrimSpace(c.Driver.Mode)) + switch mode { + case CSIModeAll, CSIModeController, CSIModeNode: + c.Driver.Mode = mode + default: + return fmt.Errorf("driver.mode must be one of: all, controller, node") + } + + c.Client.ServerURL = strings.TrimRight(strings.TrimSpace(c.Client.ServerURL), "/") + if c.Client.TimeoutSeconds <= 0 { + c.Client.TimeoutSeconds = 30 + } + if strings.TrimSpace(c.Client.AccessToken) == "" && strings.TrimSpace(c.Client.Token) != "" { + c.Client.AccessToken = strings.TrimSpace(c.Client.Token) + } + + return nil +} + +func (c *ClientConf) Timeout() time.Duration { + return time.Duration(c.TimeoutSeconds) * time.Second +} + +func (c *ClientConf) BearerToken() string { + if strings.TrimSpace(c.AccessToken) != "" { + return strings.TrimSpace(c.AccessToken) + } + return strings.TrimSpace(c.Token) +} diff --git a/config/config_test.go b/config/config_test.go new file mode 100644 index 0000000..c32e8ff --- /dev/null +++ b/config/config_test.go @@ -0,0 +1,94 @@ +package config + +import ( + "os" + "path/filepath" + "testing" +) + +func TestLoadNormalizesClientDefaults(t *testing.T) { + t.Parallel() + + dir := t.TempDir() + path := filepath.Join(dir, "csi.hcl") + if err := os.WriteFile(path, []byte(`client { + server_url = "http://192.168.84.10:9766" + access_token = "cluster-token" +}`), 0o600); err != nil { + t.Fatalf("write config: %v", err) + } + + cfg, err := Load(path) + if err != nil { + t.Fatalf("Load failed: %v", err) + } + + if cfg.Driver.Name != "storage.csi.kloud.team" { + t.Fatalf("unexpected driver name: %q", cfg.Driver.Name) + } + if cfg.Client.ServerURL != "http://192.168.84.10:9766" { + t.Fatalf("unexpected client server url: %q", cfg.Client.ServerURL) + } + if err := cfg.Validate(); err != nil { + t.Fatalf("Validate failed: %v", err) + } +} + +func TestValidateRequiresAccessToken(t *testing.T) { + t.Parallel() + + cfg := &Config{ + Driver: &DriverConf{}, + Client: &ClientConf{ServerURL: "http://127.0.0.1:9766"}, + } + if err := cfg.Validate(); err == nil { + t.Fatal("expected error for missing access token") + } +} + +func TestValidateAcceptsLegacyTokenField(t *testing.T) { + t.Parallel() + + cfg := &Config{ + Driver: &DriverConf{}, + Client: &ClientConf{ + ServerURL: "http://127.0.0.1:9766", + Token: "legacy-token", + }, + } + if err := cfg.normalize(); err != nil { + t.Fatalf("normalize failed: %v", err) + } + if cfg.Client.AccessToken != "legacy-token" { + t.Fatalf("expected legacy token migration, got %q", cfg.Client.AccessToken) + } + if err := cfg.Validate(); err != nil { + t.Fatalf("Validate failed: %v", err) + } +} + +func TestApplyEnvOverrides(t *testing.T) { + t.Setenv(envCSIServerURL, "http://csi.example:9766") + t.Setenv(envCSIAccessToken, "cluster-token") + t.Setenv(envCSIDriverMode, "node") + t.Setenv(envCSINodeID, "worker-1") + + cfg := &Config{ + Driver: &DriverConf{}, + Client: &ClientConf{}, + } + ApplyEnvOverrides(cfg) + + if cfg.Client.ServerURL != "http://csi.example:9766" { + t.Fatalf("unexpected server url: %q", cfg.Client.ServerURL) + } + if cfg.Client.AccessToken != "cluster-token" { + t.Fatalf("unexpected token: %q", cfg.Client.AccessToken) + } + if cfg.Driver.Mode != CSIModeNode { + t.Fatalf("unexpected mode: %q", cfg.Driver.Mode) + } + if cfg.Driver.NodeID != "worker-1" { + t.Fatalf("unexpected node id: %q", cfg.Driver.NodeID) + } +} diff --git a/config/env.go b/config/env.go new file mode 100644 index 0000000..47d27f4 --- /dev/null +++ b/config/env.go @@ -0,0 +1,76 @@ +package config + +import ( + "os" + "strconv" + "strings" +) + +const ( + envCSIServerURL = "KKS_CSI_SERVER_URL" + envCSIAccessToken = "KKS_CSI_ACCESS_TOKEN" + envCSIDriverMode = "KKS_CSI_DRIVER_MODE" + envCSINodeID = "KKS_CSI_NODE_ID" + envCSIDriverName = "KKS_CSI_DRIVER_NAME" + envCSIDriverEndpoint = "KKS_CSI_DRIVER_ENDPOINT" +) + +// ApplyEnvOverrides applies environment variables used by the Helm chart. +func ApplyEnvOverrides(cfg *Config) { + if cfg == nil { + return + } + if cfg.Driver == nil { + cfg.Driver = &DriverConf{} + } + if cfg.Client == nil { + cfg.Client = &ClientConf{} + } + + if v := strings.TrimSpace(os.Getenv(envCSIServerURL)); v != "" { + cfg.Client.ServerURL = strings.TrimRight(v, "/") + } + if v := strings.TrimSpace(os.Getenv(envCSIAccessToken)); v != "" { + cfg.Client.AccessToken = v + } + if v := strings.TrimSpace(os.Getenv(envCSIDriverMode)); v != "" { + cfg.Driver.Mode = strings.ToLower(v) + } + if v := strings.TrimSpace(os.Getenv(envCSINodeID)); v != "" { + cfg.Driver.NodeID = v + } + if v := strings.TrimSpace(os.Getenv(envCSIDriverName)); v != "" { + cfg.Driver.Name = v + } + if v := strings.TrimSpace(os.Getenv(envCSIDriverEndpoint)); v != "" { + cfg.Driver.Endpoint = v + } + + if cfg.Client.TimeoutSeconds <= 0 { + cfg.Client.TimeoutSeconds = 30 + } + if cfg.Driver.Mode == "" { + cfg.Driver.Mode = CSIModeNode + } + if cfg.Driver.Name == "" { + cfg.Driver.Name = "storage.csi.kloud.team" + } + if cfg.Driver.Endpoint == "" { + cfg.Driver.Endpoint = "unix:///var/lib/kubelet/plugins/storage.csi.kloud.team/csi.sock" + } +} + +func ApplyClientTimeoutFromEnv(cfg *Config) { + if cfg == nil || cfg.Client == nil { + return + } + raw := strings.TrimSpace(os.Getenv("KKS_CSI_CLIENT_TIMEOUT_SECONDS")) + if raw == "" { + return + } + seconds, err := strconv.Atoi(raw) + if err != nil || seconds <= 0 { + return + } + cfg.Client.TimeoutSeconds = seconds +} diff --git a/examples/csi-client.hcl b/examples/csi-client.hcl new file mode 100644 index 0000000..58527bc --- /dev/null +++ b/examples/csi-client.hcl @@ -0,0 +1,14 @@ +# Run on user cluster nodes via Helm, or standalone with: +# kks-csi -c examples/csi-client.hcl +# +# When installed with charts/kloud-csi, settings come from env vars instead. +driver { + name = "storage.csi.kloud.team" + endpoint = "unix:///var/lib/kubelet/plugins/storage.csi.kloud.team/csi.sock" + mode = "all" +} + +client { + server_url = "https://csi.kloud.team" + access_token = "REPLACE_WITH_CLUSTER_ACCESS_TOKEN" +} diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..db7cb4f --- /dev/null +++ b/go.mod @@ -0,0 +1,38 @@ +module github.com/KubelanCloud/kks-csi-plugin + +go 1.24.5 + +require ( + github.com/container-storage-interface/spec v1.9.0 + github.com/hashicorp/hcl/v2 v2.24.0 + github.com/spf13/cobra v1.9.1 + go.uber.org/zap v1.27.0 + google.golang.org/grpc v1.68.1 + google.golang.org/protobuf v1.34.2 + k8s.io/mount-utils v0.33.3 + k8s.io/utils v0.0.0-20241104100929-3ea5e8cea738 +) + +require ( + github.com/agext/levenshtein v1.2.1 // indirect + github.com/apparentlymart/go-textseg/v15 v15.0.0 // indirect + github.com/go-logr/logr v1.4.2 // indirect + github.com/golang/protobuf v1.5.4 // indirect + github.com/google/go-cmp v0.6.0 // indirect + github.com/inconshreveable/mousetrap v1.1.0 // indirect + github.com/mitchellh/go-wordwrap v1.0.1 // indirect + github.com/moby/sys/mountinfo v0.7.2 // indirect + github.com/spf13/pflag v1.0.6 // indirect + github.com/zclconf/go-cty v1.16.3 // indirect + go.uber.org/multierr v1.10.0 // indirect + golang.org/x/mod v0.17.0 // indirect + golang.org/x/net v0.29.0 // indirect + golang.org/x/sync v0.14.0 // indirect + golang.org/x/sys v0.33.0 // indirect + golang.org/x/text v0.25.0 // indirect + golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20240903143218-8af14fe29dc1 // indirect + k8s.io/klog/v2 v2.130.1 // indirect +) + +replace k8s.io/kubernetes => github.com/kubernetes/kubernetes v1.33.3 diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..a9d4bcf --- /dev/null +++ b/go.sum @@ -0,0 +1,73 @@ +github.com/agext/levenshtein v1.2.1 h1:QmvMAjj2aEICytGiWzmxoE0x2KZvE0fvmqMOfy2tjT8= +github.com/agext/levenshtein v1.2.1/go.mod h1:JEDfjyjHDjOF/1e4FlBE/PkbqA9OfWu2ki2W0IB5558= +github.com/apparentlymart/go-textseg/v15 v15.0.0 h1:uYvfpb3DyLSCGWnctWKGj857c6ew1u1fNQOlOtuGxQY= +github.com/apparentlymart/go-textseg/v15 v15.0.0/go.mod h1:K8XmNZdhEBkdlyDdvbmmsvpAG721bKi0joRfFdHIWJ4= +github.com/container-storage-interface/spec v1.9.0 h1:zKtX4STsq31Knz3gciCYCi1SXtO2HJDecIjDVboYavY= +github.com/container-storage-interface/spec v1.9.0/go.mod h1:ZfDu+3ZRyeVqxZM0Ds19MVLkN2d1XJ5MAfi1L3VjlT0= +github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6NIQQ7OS05n1F4g= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/go-logr/logr v1.4.2 h1:6pFjapn8bFcIbiKo3XT4j/BhANplGihG6tvd+8rYgrY= +github.com/go-logr/logr v1.4.2/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= +github.com/go-test/deep v1.0.3 h1:ZrJSEWsXzPOxaZnFteGEfooLba+ju3FYIbOrS+rQd68= +github.com/go-test/deep v1.0.3/go.mod h1:wGDj63lr65AM2AQyKZd/NYHGb0R+1RLqB8NKt3aSFNA= +github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= +github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps= +github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI= +github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= +github.com/hashicorp/hcl/v2 v2.24.0 h1:2QJdZ454DSsYGoaE6QheQZjtKZSUs9Nh2izTWiwQxvE= +github.com/hashicorp/hcl/v2 v2.24.0/go.mod h1:oGoO1FIQYfn/AgyOhlg9qLC6/nOJPX3qGbkZpYAcqfM= +github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8= +github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw= +github.com/mitchellh/go-wordwrap v1.0.1 h1:TLuKupo69TCn6TQSyGxwI1EblZZEsQ0vMlAFQflz0v0= +github.com/mitchellh/go-wordwrap v1.0.1/go.mod h1:R62XHJLzvMFRBbcrT7m7WgmE1eOyTSsCt+hzestvNj0= +github.com/moby/sys/mountinfo v0.7.2 h1:1shs6aH5s4o5H2zQLn796ADW1wMrIwHsyJ2v9KouLrg= +github.com/moby/sys/mountinfo v0.7.2/go.mod h1:1YOa8w8Ih7uW0wALDUgT1dTTSBrZ+HiBLGws92L2RU4= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM= +github.com/spf13/cobra v1.9.1 h1:CXSaggrXdbHK9CF+8ywj8Amf7PBRmPCOJugH954Nnlo= +github.com/spf13/cobra v1.9.1/go.mod h1:nDyEzZ8ogv936Cinf6g1RU9MRY64Ir93oCnqb9wxYW0= +github.com/spf13/pflag v1.0.6 h1:jFzHGLGAlb3ruxLB8MhbI6A8+AQX/2eW4qeyNZXNp2o= +github.com/spf13/pflag v1.0.6/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg= +github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA= +github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= +github.com/zclconf/go-cty v1.16.3 h1:osr++gw2T61A8KVYHoQiFbFd1Lh3JOCXc/jFLJXKTxk= +github.com/zclconf/go-cty v1.16.3/go.mod h1:VvMs5i0vgZdhYawQNq5kePSpLAoz8u1xvZgrPIxfnZE= +github.com/zclconf/go-cty-debug v0.0.0-20240509010212-0d6042c53940 h1:4r45xpDWB6ZMSMNJFMOjqrGHynW3DIBuR2H9j0ug+Mo= +github.com/zclconf/go-cty-debug v0.0.0-20240509010212-0d6042c53940/go.mod h1:CmBdvvj3nqzfzJ6nTCIwDTPZ56aVGvDrmztiO5g3qrM= +go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= +go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= +go.uber.org/multierr v1.10.0 h1:S0h4aNzvfcFsC3dRF1jLoaov7oRaKqRGC/pUEJ2yvPQ= +go.uber.org/multierr v1.10.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y= +go.uber.org/zap v1.27.0 h1:aJMhYGrd5QSmlpLMr2MftRKl7t8J8PTZPA732ud/XR8= +go.uber.org/zap v1.27.0/go.mod h1:GB2qFLM7cTU87MWRP2mPIjqfIDnGu+VIO4V/SdhGo2E= +golang.org/x/exp v0.0.0-20240719175910-8a7402abbf56 h1:2dVuKD2vS7b0QIHQbpyTISPd0LeHDbnYEryqj5Q1ug8= +golang.org/x/exp v0.0.0-20240719175910-8a7402abbf56/go.mod h1:M4RDyNAINzryxdtnbRXRL/OHtkFuWGRjvuhBJpk2IlY= +golang.org/x/mod v0.17.0 h1:zY54UmvipHiNd+pm+m0x9KhZ9hl1/7QNMyxXbc6ICqA= +golang.org/x/mod v0.17.0/go.mod h1:hTbmBsO62+eylJbnUtE2MGJUyE7QWk4xUqPFrRgJ+7c= +golang.org/x/net v0.29.0 h1:5ORfpBpCs4HzDYoodCDBbwHzdR5UrLBZ3sOnUJmFoHo= +golang.org/x/net v0.29.0/go.mod h1:gLkgy8jTGERgjzMic6DS9+SP0ajcu6Xu3Orq/SpETg0= +golang.org/x/sync v0.14.0 h1:woo0S4Yywslg6hp4eUFjTVOyKt0RookbpAHG4c1HmhQ= +golang.org/x/sync v0.14.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA= +golang.org/x/sys v0.33.0 h1:q3i8TbbEz+JRD9ywIRlyRAQbM0qF7hu24q3teo2hbuw= +golang.org/x/sys v0.33.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k= +golang.org/x/text v0.25.0 h1:qVyWApTSYLk/drJRO5mDlNYskwQznZmkpV2c8q9zls4= +golang.org/x/text v0.25.0/go.mod h1:WEdwpYrmk1qmdHvhkSTNPm3app7v4rsT8F2UD6+VHIA= +golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d h1:vU5i/LfpvrRCpgM/VPfJLg5KjxD3E+hfT1SH+d9zLwg= +golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d/go.mod h1:aiJjzUbINMkxbQROHiO6hDPo2LHcIPhhQsa9DLh0yGk= +google.golang.org/genproto/googleapis/rpc v0.0.0-20240903143218-8af14fe29dc1 h1:pPJltXNxVzT4pK9yD8vR9X75DaWYYmLGMsEvBfFQZzQ= +google.golang.org/genproto/googleapis/rpc v0.0.0-20240903143218-8af14fe29dc1/go.mod h1:UqMtugtsSgubUsoxbuAoiCXvqvErP7Gf0so0mK9tHxU= +google.golang.org/grpc v1.68.1 h1:oI5oTa11+ng8r8XMMN7jAOmWfPZWbYpCFaMUTACxkM0= +google.golang.org/grpc v1.68.1/go.mod h1:+q1XYFJjShcqn0QZHvCyeR4CXPA+llXIeUIfIe00waw= +google.golang.org/protobuf v1.34.2 h1:6xV6lTsCfpGD21XK49h7MhtcApnLqkfYgPcdHftf6hg= +google.golang.org/protobuf v1.34.2/go.mod h1:qYOHts0dSfpeUzUFpOMr/WGzszTmLH+DiWniOlNbLDw= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +k8s.io/klog/v2 v2.130.1 h1:n9Xl7H1Xvksem4KFG4PYbdQCQxqc/tTUyrgXaOhHSzk= +k8s.io/klog/v2 v2.130.1/go.mod h1:3Jpz1GvMt720eyJH1ckRHK1EDfpxISzJ7I9OYgaDtPE= +k8s.io/mount-utils v0.33.3 h1:Q1jsnqdS4LdtJSYSXgiQv/XNrRHQncLk3gMYjKNSZrE= +k8s.io/mount-utils v0.33.3/go.mod h1:1JR4rKymg8B8bCPo618hpSAdrpO6XLh0Acqok/xVwPE= +k8s.io/utils v0.0.0-20241104100929-3ea5e8cea738 h1:M3sRQVHv7vB20Xc2ybTt7ODCeFj6JSWYFzOFnYeS6Ro= +k8s.io/utils v0.0.0-20241104100929-3ea5e8cea738/go.mod h1:OLgZIPagt7ERELqWJFomSt595RzquPNLL48iOWgYOg0= diff --git a/main.go b/main.go new file mode 100644 index 0000000..9f07e57 --- /dev/null +++ b/main.go @@ -0,0 +1,66 @@ +package main + +import ( + "context" + "fmt" + "os" + "os/signal" + "syscall" + + "github.com/KubelanCloud/kks-csi-plugin/config" + "github.com/KubelanCloud/kks-csi-plugin/pkg/csi/driver" + "github.com/spf13/cobra" + "go.uber.org/zap" +) + +func main() { + logger, err := zap.NewProduction() + if err != nil { + fmt.Fprintf(os.Stderr, "init logger: %v\n", err) + os.Exit(1) + } + defer logger.Sync() //nolint:errcheck + + configPath := "csi.hcl" + + rootCmd := &cobra.Command{ + Use: "kks-csi", + Short: "Kloud CSI driver for user cluster nodes", + Long: "Runs the in-cluster CSI gRPC driver. Storage operations are sent to the kks management CSI server.", + RunE: func(cmd *cobra.Command, args []string) error { + return run(cmd, configPath, logger) + }, + } + rootCmd.PersistentFlags().StringVarP(&configPath, "config-file", "c", "csi.hcl", "Path to driver config (optional when using env vars)") + + if err := rootCmd.Execute(); err != nil { + os.Exit(1) + } +} + +func run(cmd *cobra.Command, configPath string, logger *zap.Logger) error { + cfg, err := config.LoadClient(configPath) + if err != nil { + return fmt.Errorf("load config: %w", err) + } + + ctx, cancel := signalContext(cmd.Context()) + defer cancel() + + logger.Sugar().Infof("loaded csi driver config from %s", configPath) + return driver.Run(ctx, cfg, logger) +} + +func signalContext(parent context.Context) (context.Context, context.CancelFunc) { + ctx, cancel := context.WithCancel(parent) + go func() { + ch := make(chan os.Signal, 1) + signal.Notify(ch, syscall.SIGINT, syscall.SIGTERM) + select { + case <-ch: + cancel() + case <-ctx.Done(): + } + }() + return ctx, cancel +} diff --git a/pkg/csi/api/client.go b/pkg/csi/api/client.go new file mode 100644 index 0000000..4404aa1 --- /dev/null +++ b/pkg/csi/api/client.go @@ -0,0 +1,166 @@ +package api + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "strings" + "time" + + "github.com/KubelanCloud/kks-csi-plugin/pkg/csi/provisioner" +) + +type ClientConfig struct { + BaseURL string + Token string + Timeout time.Duration +} + +type Client struct { + baseURL string + token string + client *http.Client +} + +func NewClient(cfg ClientConfig) *Client { + timeout := cfg.Timeout + if timeout <= 0 { + timeout = 30 * time.Second + } + return &Client{ + baseURL: strings.TrimRight(strings.TrimSpace(cfg.BaseURL), "/"), + token: strings.TrimSpace(cfg.Token), + client: &http.Client{Timeout: timeout}, + } +} + +func (c *Client) Close() error { + return nil +} + +func (c *Client) ClusterInfo(ctx context.Context) (provisioner.ClusterInfo, error) { + var out provisioner.ClusterInfo + if err := c.doJSON(ctx, http.MethodGet, "/v1/cluster", nil, &out); err != nil { + return provisioner.ClusterInfo{}, err + } + return out, nil +} + +func (c *Client) CreateVolume(ctx context.Context, req provisioner.CreateVolumeRequest) (provisioner.Volume, error) { + var out provisioner.Volume + if err := c.doJSON(ctx, http.MethodPost, "/v1/volumes", req, &out); err != nil { + return provisioner.Volume{}, err + } + return out, nil +} + +func (c *Client) DeleteVolume(ctx context.Context, volumeID string) error { + return c.doJSON(ctx, http.MethodDelete, "/v1/volumes/"+escapePath(volumeID), nil, nil) +} + +func (c *Client) VolumeExists(ctx context.Context, volumeID string) (bool, error) { + var out provisioner.VolumeExistsResponse + err := c.doJSON(ctx, http.MethodGet, "/v1/volumes/"+escapePath(volumeID), nil, &out) + if err != nil { + if isNotFound(err) { + return false, nil + } + return false, err + } + return out.Exists, nil +} + +func (c *Client) PublishVolume(ctx context.Context, volumeID, nodeID string) (provisioner.PublishVolumeResponse, error) { + var out provisioner.PublishVolumeResponse + req := provisioner.PublishVolumeRequest{NodeID: nodeID} + path := "/v1/volumes/" + escapePath(volumeID) + "/publish" + if err := c.doJSON(ctx, http.MethodPost, path, req, &out); err != nil { + return provisioner.PublishVolumeResponse{}, err + } + return out, nil +} + +func (c *Client) UnpublishVolume(ctx context.Context, volumeID, nodeID string) error { + req := provisioner.UnpublishVolumeRequest{ + NodeID: nodeID, + VolumeID: volumeID, + } + return c.doJSON(ctx, http.MethodPost, "/v1/volumes/unpublish", req, nil) +} + +func (c *Client) doJSON(ctx context.Context, method, path string, reqBody any, respBody any) error { + var body io.Reader + if reqBody != nil { + raw, err := json.Marshal(reqBody) + if err != nil { + return fmt.Errorf("marshal request: %w", err) + } + body = bytes.NewReader(raw) + } + + req, err := http.NewRequestWithContext(ctx, method, c.baseURL+path, body) + if err != nil { + return fmt.Errorf("create request: %w", err) + } + req.Header.Set("Accept", "application/json") + if reqBody != nil { + req.Header.Set("Content-Type", "application/json") + } + if c.token != "" { + req.Header.Set("Authorization", "Bearer "+c.token) + } + + resp, err := c.client.Do(req) + if err != nil { + return fmt.Errorf("request %s %s: %w", method, path, err) + } + defer resp.Body.Close() + + raw, err := io.ReadAll(resp.Body) + if err != nil { + return fmt.Errorf("read response: %w", err) + } + + if resp.StatusCode >= 400 { + return &HTTPError{ + StatusCode: resp.StatusCode, + Method: method, + Path: path, + Body: strings.TrimSpace(string(raw)), + } + } + + if respBody == nil || len(raw) == 0 { + return nil + } + if err := json.Unmarshal(raw, respBody); err != nil { + return fmt.Errorf("decode response: %w", err) + } + return nil +} + +type HTTPError struct { + StatusCode int + Method string + Path string + Body string +} + +func (e *HTTPError) Error() string { + if e.Body == "" { + return fmt.Sprintf("%s %s: status %d", e.Method, e.Path, e.StatusCode) + } + return fmt.Sprintf("%s %s: status %d: %s", e.Method, e.Path, e.StatusCode, e.Body) +} + +func isNotFound(err error) bool { + httpErr, ok := err.(*HTTPError) + return ok && httpErr.StatusCode == http.StatusNotFound +} + +func escapePath(value string) string { + return strings.ReplaceAll(value, "/", "%2F") +} diff --git a/pkg/csi/api/client_test.go b/pkg/csi/api/client_test.go new file mode 100644 index 0000000..bf09590 --- /dev/null +++ b/pkg/csi/api/client_test.go @@ -0,0 +1,62 @@ +package api + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + "github.com/KubelanCloud/kks-csi-plugin/pkg/csi/provisioner" +) + +func TestClientClusterInfo(t *testing.T) { + t.Parallel() + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/v1/cluster" { + t.Fatalf("unexpected path: %s", r.URL.Path) + } + _ = json.NewEncoder(w).Encode(provisioner.ClusterInfo{ + StorageID: "abc123", + }) + })) + t.Cleanup(srv.Close) + + client := NewClient(ClientConfig{BaseURL: srv.URL}) + info, err := client.ClusterInfo(context.Background()) + if err != nil { + t.Fatalf("ClusterInfo failed: %v", err) + } + if info.StorageID != "abc123" { + t.Fatalf("unexpected cluster info: %#v", info) + } +} + +func TestClientCreateVolume(t *testing.T) { + t.Parallel() + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost || r.URL.Path != "/v1/volumes" { + t.Fatalf("unexpected request: %s %s", r.Method, r.URL.Path) + } + w.WriteHeader(http.StatusCreated) + _ = json.NewEncoder(w).Encode(provisioner.Volume{ + VolumeID: "abc123/k8s-volumes/pvc-1", + SizeBytes: 1024, + }) + })) + t.Cleanup(srv.Close) + + client := NewClient(ClientConfig{BaseURL: srv.URL}) + vol, err := client.CreateVolume(context.Background(), provisioner.CreateVolumeRequest{ + Name: "pvc-1", + SizeBytes: 1024, + }) + if err != nil { + t.Fatalf("CreateVolume failed: %v", err) + } + if vol.VolumeID != "abc123/k8s-volumes/pvc-1" { + t.Fatalf("unexpected volume: %#v", vol) + } +} diff --git a/pkg/csi/client/backend.go b/pkg/csi/client/backend.go new file mode 100644 index 0000000..34a78e9 --- /dev/null +++ b/pkg/csi/client/backend.go @@ -0,0 +1,15 @@ +package client + +import ( + "github.com/KubelanCloud/kks-csi-plugin/config" + "github.com/KubelanCloud/kks-csi-plugin/pkg/csi/api" + "github.com/KubelanCloud/kks-csi-plugin/pkg/csi/provisioner" +) + +func NewBackend(cfg *config.ClientConf) provisioner.Backend { + return api.NewClient(api.ClientConfig{ + BaseURL: cfg.ServerURL, + Token: cfg.BearerToken(), + Timeout: cfg.Timeout(), + }) +} diff --git a/pkg/csi/driver/controller.go b/pkg/csi/driver/controller.go new file mode 100644 index 0000000..7fec546 --- /dev/null +++ b/pkg/csi/driver/controller.go @@ -0,0 +1,168 @@ +package driver + +import ( + "context" + + "github.com/container-storage-interface/spec/lib/go/csi" + "github.com/KubelanCloud/kks-csi-plugin/pkg/csi/provisioner" +) + +type ControllerServer struct { + d *Driver +} + +func newControllerServer(d *Driver) *ControllerServer { + return &ControllerServer{d: d} +} + +func (s *ControllerServer) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest) (*csi.CreateVolumeResponse, error) { + if req.GetName() == "" { + return nil, invalidArgument("volume name is required") + } + if req.GetVolumeCapabilities() == nil { + return nil, invalidArgument("volume capabilities are required") + } + + capacity := int64(1 * 1024 * 1024 * 1024) + if req.GetCapacityRange() != nil { + if req.GetCapacityRange().GetRequiredBytes() > 0 { + capacity = req.GetCapacityRange().GetRequiredBytes() + } else if req.GetCapacityRange().GetLimitBytes() > 0 { + capacity = req.GetCapacityRange().GetLimitBytes() + } + } + + vol, err := s.d.backend.CreateVolume(ctx, provisioner.CreateVolumeRequest{ + Name: sanitizeVolumeName(req.GetName()), + SizeBytes: capacity, + }) + if err != nil { + return nil, internalError(err) + } + + return &csi.CreateVolumeResponse{ + Volume: &csi.Volume{ + VolumeId: vol.VolumeID, + CapacityBytes: vol.SizeBytes, + VolumeContext: vol.VolumeContext, + }, + }, nil +} + +func (s *ControllerServer) DeleteVolume(ctx context.Context, req *csi.DeleteVolumeRequest) (*csi.DeleteVolumeResponse, error) { + if req.GetVolumeId() == "" { + return nil, invalidArgument("volume id is required") + } + + if err := s.d.backend.DeleteVolume(ctx, req.GetVolumeId()); err != nil { + return nil, internalError(err) + } + return &csi.DeleteVolumeResponse{}, nil +} + +func (s *ControllerServer) ControllerPublishVolume(ctx context.Context, req *csi.ControllerPublishVolumeRequest) (*csi.ControllerPublishVolumeResponse, error) { + if req.GetVolumeId() == "" { + return nil, invalidArgument("volume id is required") + } + if req.GetNodeId() == "" { + return nil, invalidArgument("node id is required") + } + + pub, err := s.d.backend.PublishVolume(ctx, req.GetVolumeId(), req.GetNodeId()) + if err != nil { + return nil, internalError(err) + } + + return &csi.ControllerPublishVolumeResponse{ + PublishContext: pub.PublishContext, + }, nil +} + +func (s *ControllerServer) ControllerUnpublishVolume(ctx context.Context, req *csi.ControllerUnpublishVolumeRequest) (*csi.ControllerUnpublishVolumeResponse, error) { + if req.GetVolumeId() == "" { + return nil, invalidArgument("volume id is required") + } + nodeID := req.GetNodeId() + if nodeID == "" { + return &csi.ControllerUnpublishVolumeResponse{}, nil + } + + if err := s.d.backend.UnpublishVolume(ctx, req.GetVolumeId(), nodeID); err != nil { + return nil, internalError(err) + } + return &csi.ControllerUnpublishVolumeResponse{}, nil +} + +func (s *ControllerServer) ValidateVolumeCapabilities(ctx context.Context, req *csi.ValidateVolumeCapabilitiesRequest) (*csi.ValidateVolumeCapabilitiesResponse, error) { + _ = ctx + if req.GetVolumeId() == "" { + return nil, invalidArgument("volume id is required") + } + if req.GetVolumeCapabilities() == nil { + return nil, invalidArgument("volume capabilities are required") + } + return &csi.ValidateVolumeCapabilitiesResponse{ + Confirmed: &csi.ValidateVolumeCapabilitiesResponse_Confirmed{ + VolumeCapabilities: req.GetVolumeCapabilities(), + }, + }, nil +} + +func (s *ControllerServer) ListVolumes(ctx context.Context, req *csi.ListVolumesRequest) (*csi.ListVolumesResponse, error) { + _ = ctx + _ = req + return nil, unimplemented("ListVolumes is not supported") +} + +func (s *ControllerServer) GetCapacity(ctx context.Context, req *csi.GetCapacityRequest) (*csi.GetCapacityResponse, error) { + _ = ctx + _ = req + return nil, unimplemented("GetCapacity is not supported") +} + +func (s *ControllerServer) CreateSnapshot(ctx context.Context, req *csi.CreateSnapshotRequest) (*csi.CreateSnapshotResponse, error) { + _ = ctx + _ = req + return nil, unimplemented("CreateSnapshot is not supported") +} + +func (s *ControllerServer) DeleteSnapshot(ctx context.Context, req *csi.DeleteSnapshotRequest) (*csi.DeleteSnapshotResponse, error) { + _ = ctx + _ = req + return nil, unimplemented("DeleteSnapshot is not supported") +} + +func (s *ControllerServer) ListSnapshots(ctx context.Context, req *csi.ListSnapshotsRequest) (*csi.ListSnapshotsResponse, error) { + _ = ctx + _ = req + return nil, unimplemented("ListSnapshots is not supported") +} + +func (s *ControllerServer) ControllerExpandVolume(ctx context.Context, req *csi.ControllerExpandVolumeRequest) (*csi.ControllerExpandVolumeResponse, error) { + _ = ctx + _ = req + return nil, unimplemented("ControllerExpandVolume is not supported") +} + +func (s *ControllerServer) ControllerGetVolume(ctx context.Context, req *csi.ControllerGetVolumeRequest) (*csi.ControllerGetVolumeResponse, error) { + _ = ctx + _ = req + return nil, unimplemented("ControllerGetVolume is not supported") +} + +func (s *ControllerServer) ControllerGetCapabilities(ctx context.Context, req *csi.ControllerGetCapabilitiesRequest) (*csi.ControllerGetCapabilitiesResponse, error) { + _ = ctx + _ = req + return &csi.ControllerGetCapabilitiesResponse{ + Capabilities: []*csi.ControllerServiceCapability{ + controllerCapability(csi.ControllerServiceCapability_RPC_CREATE_DELETE_VOLUME), + controllerCapability(csi.ControllerServiceCapability_RPC_PUBLISH_UNPUBLISH_VOLUME), + }, + }, nil +} + +func (s *ControllerServer) ControllerModifyVolume(ctx context.Context, req *csi.ControllerModifyVolumeRequest) (*csi.ControllerModifyVolumeResponse, error) { + _ = ctx + _ = req + return nil, unimplemented("ControllerModifyVolume is not supported") +} diff --git a/pkg/csi/driver/driver.go b/pkg/csi/driver/driver.go new file mode 100644 index 0000000..3f3225a --- /dev/null +++ b/pkg/csi/driver/driver.go @@ -0,0 +1,149 @@ +package driver + +import ( + "context" + "fmt" + "net" + "os" + "path/filepath" + "strings" + + "github.com/container-storage-interface/spec/lib/go/csi" + "github.com/KubelanCloud/kks-csi-plugin/config" + "github.com/KubelanCloud/kks-csi-plugin/pkg/csi/client" + "github.com/KubelanCloud/kks-csi-plugin/pkg/csi/provisioner" + "go.uber.org/zap" + "google.golang.org/grpc" +) + +type Driver struct { + cfg *config.Config + log *zap.SugaredLogger + backend provisioner.Backend + storageID string + identity *IdentityServer + controller *ControllerServer + node *NodeServer +} + +func Run(ctx context.Context, cfg *config.Config, logger *zap.Logger) error { + if err := cfg.Validate(); err != nil { + return err + } + if logger == nil { + logger = zap.NewNop() + } + log := logger.Sugar() + + backend := client.NewBackend(cfg.Client) + defer backend.Close() + + cluster, err := backend.ClusterInfo(ctx) + if err != nil { + return fmt.Errorf("resolve storage info from csi server: %w", err) + } + + d := &Driver{ + cfg: cfg, + log: log, + backend: backend, + storageID: cluster.StorageID, + identity: newIdentityServer(cfg), + } + + switch cfg.Driver.Mode { + case config.CSIModeController, config.CSIModeAll: + d.controller = newControllerServer(d) + case config.CSIModeNode: + } + + switch cfg.Driver.Mode { + case config.CSIModeNode, config.CSIModeAll: + d.node = newNodeServer(d) + case config.CSIModeController: + } + + endpoint, err := parseEndpoint(cfg.Driver.Endpoint) + if err != nil { + return err + } + if err := ensureSocketDir(endpoint); err != nil { + return err + } + + listener, err := net.Listen(endpoint.network, endpoint.address) + if err != nil { + return fmt.Errorf("listen on %s: %w", cfg.Driver.Endpoint, err) + } + defer listener.Close() + + server := grpc.NewServer() + csi.RegisterIdentityServer(server, d.identity) + if d.controller != nil { + csi.RegisterControllerServer(server, d.controller) + } + if d.node != nil { + csi.RegisterNodeServer(server, d.node) + } + + log.Infof( + "starting csi client name=%s endpoint=%s mode=%s node_id=%s server=%s storage_id=%s", + cfg.Driver.Name, + cfg.Driver.Endpoint, + cfg.Driver.Mode, + cfg.Driver.NodeID, + cfg.Client.ServerURL, + cluster.StorageID, + ) + + errCh := make(chan error, 1) + go func() { + errCh <- server.Serve(listener) + }() + + select { + case <-ctx.Done(): + log.Info("shutting down csi client") + server.GracefulStop() + return nil + case err := <-errCh: + if err != nil { + return fmt.Errorf("csi grpc server stopped: %w", err) + } + return nil + } +} + +type endpoint struct { + network string + address string +} + +func parseEndpoint(raw string) (endpoint, error) { + if strings.HasPrefix(raw, "unix://") { + return endpoint{ + network: "unix", + address: strings.TrimPrefix(raw, "unix://"), + }, nil + } + if strings.HasPrefix(raw, "tcp://") { + return endpoint{ + network: "tcp", + address: strings.TrimPrefix(raw, "tcp://"), + }, nil + } + return endpoint{}, fmt.Errorf("unsupported endpoint %q (use unix:// or tcp://)", raw) +} + +func ensureSocketDir(ep endpoint) error { + if ep.network != "unix" { + return nil + } + if err := os.MkdirAll(filepath.Dir(ep.address), 0o755); err != nil { + return fmt.Errorf("create socket directory: %w", err) + } + if err := os.Remove(ep.address); err != nil && !os.IsNotExist(err) { + return fmt.Errorf("remove stale socket: %w", err) + } + return nil +} diff --git a/pkg/csi/driver/identity.go b/pkg/csi/driver/identity.go new file mode 100644 index 0000000..932d5e8 --- /dev/null +++ b/pkg/csi/driver/identity.go @@ -0,0 +1,82 @@ +package driver + +import ( + "context" + + "github.com/container-storage-interface/spec/lib/go/csi" + "github.com/KubelanCloud/kks-csi-plugin/config" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + "google.golang.org/protobuf/types/known/wrapperspb" +) + +type IdentityServer struct { + cfg *config.Config +} + +func newIdentityServer(cfg *config.Config) *IdentityServer { + return &IdentityServer{cfg: cfg} +} + +func (s *IdentityServer) GetPluginInfo(ctx context.Context, req *csi.GetPluginInfoRequest) (*csi.GetPluginInfoResponse, error) { + _ = ctx + _ = req + return &csi.GetPluginInfoResponse{ + Name: s.cfg.Driver.Name, + VendorVersion: "v0.1.0", + }, nil +} + +func (s *IdentityServer) GetPluginCapabilities(ctx context.Context, req *csi.GetPluginCapabilitiesRequest) (*csi.GetPluginCapabilitiesResponse, error) { + _ = ctx + _ = req + return &csi.GetPluginCapabilitiesResponse{ + Capabilities: []*csi.PluginCapability{ + { + Type: &csi.PluginCapability_Service_{ + Service: &csi.PluginCapability_Service{ + Type: csi.PluginCapability_Service_CONTROLLER_SERVICE, + }, + }, + }, + }, + }, nil +} + +func (s *IdentityServer) Probe(ctx context.Context, req *csi.ProbeRequest) (*csi.ProbeResponse, error) { + _ = ctx + _ = req + return &csi.ProbeResponse{Ready: wrapperspb.Bool(true)}, nil +} + +func controllerCapability(t csi.ControllerServiceCapability_RPC_Type) *csi.ControllerServiceCapability { + return &csi.ControllerServiceCapability{ + Type: &csi.ControllerServiceCapability_Rpc{ + Rpc: &csi.ControllerServiceCapability_RPC{Type: t}, + }, + } +} + +func nodeCapability(t csi.NodeServiceCapability_RPC_Type) *csi.NodeServiceCapability { + return &csi.NodeServiceCapability{ + Type: &csi.NodeServiceCapability_Rpc{ + Rpc: &csi.NodeServiceCapability_RPC{Type: t}, + }, + } +} + +func invalidArgument(msg string) error { + return status.Error(codes.InvalidArgument, msg) +} + +func notFound(msg string) error { + return status.Error(codes.NotFound, msg) +} + +func internalError(err error) error { + return status.Errorf(codes.Internal, "%v", err) +} + +func unimplemented(msg string) error { + return status.Error(codes.Unimplemented, msg) +} diff --git a/pkg/csi/driver/mount_linux.go b/pkg/csi/driver/mount_linux.go new file mode 100644 index 0000000..538a388 --- /dev/null +++ b/pkg/csi/driver/mount_linux.go @@ -0,0 +1,120 @@ +//go:build linux + +package driver + +import ( + "fmt" + "os" + "os/exec" + "path/filepath" + "strconv" + "strings" + + mount "k8s.io/mount-utils" + utilexec "k8s.io/utils/exec" +) + +const defaultLinuxFsType = "ext4" + +func newMounter() *mount.SafeFormatAndMount { + return mount.NewSafeFormatAndMount(mount.New(""), utilexec.New()) +} + +func scsiHostRescan() { + scsiPath := "/sys/class/scsi_host/" + entries, err := os.ReadDir(scsiPath) + if err != nil { + return + } + for _, entry := range entries { + name := filepath.Join(scsiPath, entry.Name(), "scan") + _ = os.WriteFile(name, []byte("- - -"), 0o666) + } +} + +func findDiskByLUN(lun int) (string, error) { + scsiHostRescan() + + sysPath := "/sys/bus/scsi/devices" + entries, err := os.ReadDir(sysPath) + if err != nil { + return "", fmt.Errorf("read %s: %w", sysPath, err) + } + + for _, entry := range entries { + name := entry.Name() + parts := strings.Split(name, ":") + if len(parts) < 4 { + continue + } + deviceLUN, err := strconv.Atoi(parts[3]) + if err != nil || deviceLUN != lun { + continue + } + + vendorPath := filepath.Join(sysPath, name, "vendor") + vendorBytes, err := os.ReadFile(vendorPath) + if err != nil { + continue + } + vendor := strings.TrimSpace(string(vendorBytes)) + if vendor != "QEMU" && strings.ToUpper(vendor) != "MSFT" { + continue + } + + blockDir := filepath.Join(sysPath, name, "block") + devices, err := os.ReadDir(blockDir) + if err != nil || len(devices) == 0 { + continue + } + devName := devices[0].Name() + + for _, devLinkPath := range []string{"/dev/disk/by-id/", "/dev/disk/by-path/"} { + if link, err := findDiskLink(devLinkPath, devName); err == nil { + return link, nil + } + } + return "/dev/" + devName, nil + } + + return "", fmt.Errorf("failed to find disk by lun %d", lun) +} + +func findDiskLink(devLinkPath, devName string) (string, error) { + entries, err := os.ReadDir(devLinkPath) + if err != nil { + return "", err + } + for _, entry := range entries { + linkPath := filepath.Join(devLinkPath, entry.Name()) + target, err := os.Readlink(linkPath) + if err != nil { + continue + } + if strings.HasSuffix(target, devName) { + return linkPath, nil + } + } + return "", fmt.Errorf("device %s not found under %s", devName, devLinkPath) +} + +func formatAndMount(source, target, fsType string, options []string, mounter *mount.SafeFormatAndMount) error { + return mounter.FormatAndMount(source, target, fsType, options) +} + +func cleanupMountPoint(target string, mounter *mount.SafeFormatAndMount) error { + return mount.CleanupMountPoint(target, mounter, true) +} + +func bindMount(source, target string) error { + out, err := exec.Command("mount", "--bind", source, target).CombinedOutput() + if err != nil { + return fmt.Errorf("bind mount %s -> %s: %w: %s", source, target, err, strings.TrimSpace(string(out))) + } + return nil +} + +func isMounted(path string) bool { + out, err := exec.Command("findmnt", "-n", path).CombinedOutput() + return err == nil && strings.TrimSpace(string(out)) != "" +} diff --git a/pkg/csi/driver/mount_other.go b/pkg/csi/driver/mount_other.go new file mode 100644 index 0000000..7470aae --- /dev/null +++ b/pkg/csi/driver/mount_other.go @@ -0,0 +1,39 @@ +//go:build !linux + +package driver + +import ( + "fmt" + + mount "k8s.io/mount-utils" + utilexec "k8s.io/utils/exec" +) + +const defaultLinuxFsType = "ext4" + +func newMounter() *mount.SafeFormatAndMount { + return mount.NewSafeFormatAndMount(mount.New(""), utilexec.New()) +} + +func findDiskByLUN(lun int) (string, error) { + return "", fmt.Errorf("scsi volume mount is only supported on linux (lun %d)", lun) +} + +func formatAndMount(source, target, fsType string, options []string, mounter *mount.SafeFormatAndMount) error { + return mounter.FormatAndMount(source, target, fsType, options) +} + +func cleanupMountPoint(target string, mounter *mount.SafeFormatAndMount) error { + return mount.CleanupMountPoint(target, mounter, true) +} + +func bindMount(source, target string) error { + m := newMounter() + return m.Mount(source, target, "", []string{"bind"}) +} + +func isMounted(path string) bool { + m := newMounter() + notMnt, err := m.IsLikelyNotMountPoint(path) + return err == nil && !notMnt +} diff --git a/pkg/csi/driver/node.go b/pkg/csi/driver/node.go new file mode 100644 index 0000000..c0a7e52 --- /dev/null +++ b/pkg/csi/driver/node.go @@ -0,0 +1,165 @@ +package driver + +import ( + "context" + "fmt" + "os" + "path/filepath" + "strconv" + "strings" + + "github.com/container-storage-interface/spec/lib/go/csi" + "github.com/KubelanCloud/kks-csi-plugin/pkg/csi/provisioner" +) + +type NodeServer struct { + d *Driver +} + +func newNodeServer(d *Driver) *NodeServer { + return &NodeServer{d: d} +} + +func (s *NodeServer) NodeStageVolume(ctx context.Context, req *csi.NodeStageVolumeRequest) (*csi.NodeStageVolumeResponse, error) { + _ = ctx + if req.GetVolumeId() == "" { + return nil, invalidArgument("volume id is required") + } + if req.GetStagingTargetPath() == "" { + return nil, invalidArgument("staging target path is required") + } + if isBlockVolume(req.GetVolumeCapability()) { + return nil, unimplemented("block volumes are not supported") + } + + lun, err := lunFromPublishContext(req.GetPublishContext()) + if err != nil { + return nil, invalidArgument(err.Error()) + } + + device, err := findDiskByLUN(lun) + if err != nil { + return nil, internalError(err) + } + + if err := os.MkdirAll(req.GetStagingTargetPath(), 0o755); err != nil { + return nil, internalError(fmt.Errorf("create staging path: %w", err)) + } + + fsType := defaultLinuxFsType + options := []string{} + if mnt := req.GetVolumeCapability().GetMount(); mnt != nil { + if mnt.FsType != "" { + fsType = mnt.FsType + } + options = append(options, mnt.MountFlags...) + } + + mounter := newMounter() + if err := formatAndMount(device, req.GetStagingTargetPath(), fsType, options, mounter); err != nil { + return nil, internalError(fmt.Errorf("format and mount %s: %w", device, err)) + } + + return &csi.NodeStageVolumeResponse{}, nil +} + +func (s *NodeServer) NodeUnstageVolume(ctx context.Context, req *csi.NodeUnstageVolumeRequest) (*csi.NodeUnstageVolumeResponse, error) { + _ = ctx + if req.GetVolumeId() == "" { + return nil, invalidArgument("volume id is required") + } + if req.GetStagingTargetPath() == "" { + return nil, invalidArgument("staging target path is required") + } + + mounter := newMounter() + if err := cleanupMountPoint(req.GetStagingTargetPath(), mounter); err != nil { + return nil, internalError(err) + } + return &csi.NodeUnstageVolumeResponse{}, nil +} + +func (s *NodeServer) NodePublishVolume(ctx context.Context, req *csi.NodePublishVolumeRequest) (*csi.NodePublishVolumeResponse, error) { + _ = ctx + if req.GetVolumeId() == "" { + return nil, invalidArgument("volume id is required") + } + if req.GetTargetPath() == "" { + return nil, invalidArgument("target path is required") + } + if req.GetStagingTargetPath() == "" { + return nil, invalidArgument("staging target path is required") + } + + if err := os.MkdirAll(filepath.Dir(req.GetTargetPath()), 0o755); err != nil { + return nil, internalError(err) + } + + if err := bindMount(req.GetStagingTargetPath(), req.GetTargetPath()); err != nil { + return nil, internalError(err) + } + + return &csi.NodePublishVolumeResponse{}, nil +} + +func (s *NodeServer) NodeUnpublishVolume(ctx context.Context, req *csi.NodeUnpublishVolumeRequest) (*csi.NodeUnpublishVolumeResponse, error) { + _ = ctx + if req.GetVolumeId() == "" { + return nil, invalidArgument("volume id is required") + } + if req.GetTargetPath() == "" { + return nil, invalidArgument("target path is required") + } + + mounter := newMounter() + if err := cleanupMountPoint(req.GetTargetPath(), mounter); err != nil { + return nil, internalError(err) + } + return &csi.NodeUnpublishVolumeResponse{}, nil +} + +func (s *NodeServer) NodeGetVolumeStats(ctx context.Context, req *csi.NodeGetVolumeStatsRequest) (*csi.NodeGetVolumeStatsResponse, error) { + _ = ctx + _ = req + return nil, unimplemented("NodeGetVolumeStats is not supported") +} + +func (s *NodeServer) NodeExpandVolume(ctx context.Context, req *csi.NodeExpandVolumeRequest) (*csi.NodeExpandVolumeResponse, error) { + _ = ctx + _ = req + return nil, unimplemented("NodeExpandVolume is not supported") +} + +func (s *NodeServer) NodeGetCapabilities(ctx context.Context, req *csi.NodeGetCapabilitiesRequest) (*csi.NodeGetCapabilitiesResponse, error) { + _ = ctx + _ = req + return &csi.NodeGetCapabilitiesResponse{ + Capabilities: []*csi.NodeServiceCapability{ + nodeCapability(csi.NodeServiceCapability_RPC_STAGE_UNSTAGE_VOLUME), + }, + }, nil +} + +func (s *NodeServer) NodeGetInfo(ctx context.Context, req *csi.NodeGetInfoRequest) (*csi.NodeGetInfoResponse, error) { + _ = ctx + _ = req + return &csi.NodeGetInfoResponse{ + NodeId: s.d.cfg.Driver.NodeID, + }, nil +} + +func lunFromPublishContext(publishContext map[string]string) (int, error) { + lunStr := strings.TrimSpace(publishContext[provisioner.PublishContextLUN]) + if lunStr == "" { + return 0, fmt.Errorf("publish context %q is required", provisioner.PublishContextLUN) + } + lun, err := strconv.Atoi(lunStr) + if err != nil { + return 0, fmt.Errorf("invalid publish context %q: %w", provisioner.PublishContextLUN, err) + } + return lun, nil +} + +func isBlockVolume(capability *csi.VolumeCapability) bool { + return capability != nil && capability.GetBlock() != nil +} diff --git a/pkg/csi/driver/volume.go b/pkg/csi/driver/volume.go new file mode 100644 index 0000000..e2707eb --- /dev/null +++ b/pkg/csi/driver/volume.go @@ -0,0 +1,7 @@ +package driver + +import "github.com/KubelanCloud/kks-csi-plugin/pkg/csi/volume" + +func sanitizeVolumeName(name string) string { + return volume.SanitizeName(name) +} diff --git a/pkg/csi/provisioner/types.go b/pkg/csi/provisioner/types.go new file mode 100644 index 0000000..a8faf74 --- /dev/null +++ b/pkg/csi/provisioner/types.go @@ -0,0 +1,51 @@ +package provisioner + +import "context" + +const ( + PublishContextLUN = "scsi.lun" + PublishContextSlot = "proxmox.scsi_slot" + PublishContextVolid = "proxmox.volid" +) + +type ClusterInfo struct { + StorageID string `json:"storage_id"` +} + +type Volume struct { + VolumeID string `json:"volume_id"` + SizeBytes int64 `json:"size_bytes"` + VolumeContext map[string]string `json:"volume_context,omitempty"` +} + +type CreateVolumeRequest struct { + Name string `json:"name"` + SizeBytes int64 `json:"size_bytes"` +} + +type PublishVolumeRequest struct { + NodeID string `json:"node_id"` +} + +type PublishVolumeResponse struct { + PublishContext map[string]string `json:"publish_context"` +} + +type UnpublishVolumeRequest struct { + NodeID string `json:"node_id"` + VolumeID string `json:"volume_id"` +} + +type VolumeExistsResponse struct { + Exists bool `json:"exists"` +} + +type Backend interface { + ClusterInfo(ctx context.Context) (ClusterInfo, error) + CreateVolume(ctx context.Context, req CreateVolumeRequest) (Volume, error) + DeleteVolume(ctx context.Context, volumeID string) error + VolumeExists(ctx context.Context, volumeID string) (bool, error) + PublishVolume(ctx context.Context, volumeID, nodeID string) (PublishVolumeResponse, error) + UnpublishVolume(ctx context.Context, volumeID, nodeID string) error + Close() error +} diff --git a/pkg/csi/volume/id.go b/pkg/csi/volume/id.go new file mode 100644 index 0000000..61d637c --- /dev/null +++ b/pkg/csi/volume/id.go @@ -0,0 +1,29 @@ +package volume + +import ( + "fmt" + "path" + "strings" +) + +func ID(storageID, name string) string { + return path.Join(storageID, name) +} + +func Parse(volumeID string) (storageID, name string, err error) { + parts := strings.Split(volumeID, "/") + if len(parts) != 2 { + return "", "", fmt.Errorf("invalid volume id %q", volumeID) + } + return parts[0], parts[1], nil +} + +func SanitizeName(name string) string { + name = strings.TrimSpace(name) + name = strings.ReplaceAll(name, " ", "-") + return name +} + +func ExportKey(volumeID string) string { + return strings.NewReplacer("/", "_", " ", "_").Replace(volumeID) +} diff --git a/pkg/csi/volume/id_test.go b/pkg/csi/volume/id_test.go new file mode 100644 index 0000000..435c1aa --- /dev/null +++ b/pkg/csi/volume/id_test.go @@ -0,0 +1,25 @@ +package volume + +import "testing" + +func TestParseVolumeID(t *testing.T) { + storageID, name, err := Parse("nfs-prod/pvc-1") + if err != nil { + t.Fatalf("Parse failed: %v", err) + } + if storageID != "nfs-prod" || name != "pvc-1" { + t.Fatalf("unexpected parse result: %q %q", storageID, name) + } +} + +func TestParseVolumeIDRejectsInvalid(t *testing.T) { + if _, _, err := Parse("only-one-part"); err == nil { + t.Fatal("expected error for invalid volume id") + } +} + +func TestID(t *testing.T) { + if got := ID("nfs-prod", "pvc-1"); got != "nfs-prod/pvc-1" { + t.Fatalf("unexpected id: %q", got) + } +}