Initial commit

.gitignore

@@ -0,0 +1 @@
+bin/

Dockerfile

@@ -0,0 +1,4 @@
+FROM public.ecr.aws/eks-distro-build-tooling/eks-distro-minimal-base:latest.2 AS linux-amazon
+
+COPY ./main /bin/ebs-external-volume-modifier
+ENTRYPOINT ["/bin/ebs-external-volume-modifier"]

Makefile

@@ -0,0 +1,33 @@
+PROTO_FILE=modify.proto
+PROTO_GENERATED_FILES_PATH=pkg/rpc
+VERSION="v0.1.0"
+
+.PHONY: all
+all: build
+
+.PHONY: build
+build:
+	go build -o bin/main -ldflags="-X 'main.version=$(VERSION)'" cmd/main.go
+
+.PHONY: proto
+proto:
+	protoc --go_out=$(PROTO_GENERATED_FILES_PATH) --go_opt=paths=source_relative --go-grpc_out=$(PROTO_GENERATED_FILES_PATH) --go-grpc_opt=paths=source_relative $(PROTO_FILE)
+
+.PHONY: test
+test:
+	go test ./... -race
+
+.PHONY: clean
+clean:
+	rm -rf bin/
+
+
+.PHONY: check
+check: check-proto
+
+.PHONY: check-proto
+check-proto:
+	$(eval TMPDIR := $(shell mktemp -d))
+	protoc --go_out=$(TMPDIR) --go_opt=paths=source_relative --go-grpc_out=$(TMPDIR) --go-grpc_opt=paths=source_relative $(PROTO_FILE)
+	diff -qr $(TMPDIR) $(PROTO_GENERATED_FILES_PATH) || (printf "\nThe proto file seems to have been modified. PLease run `make proto`."; exit 1)
+	rm -rf $(TMPDIR)

README.md

@@ -1,11 +1,7 @@
-## My Project
+## volume-modifier-for-k8s
 
-TODO: Fill this README out!
+volume-modifier-for-k8s is a sidecar deployed alongside CSI drivers to enable volume modification through annotations on the PVC.
 
-Be sure to:
-
-* Change the title in this README
-* Edit your repository description on GitHub
 
 ## Security
 

cmd/main.go

@@ -0,0 +1,173 @@
+package main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"net/http"
+	"os"
+	"time"
+
+	csi "github.com/awslabs/volume-modifier-for-k8s/pkg/client"
+	"github.com/awslabs/volume-modifier-for-k8s/pkg/controller"
+	"github.com/awslabs/volume-modifier-for-k8s/pkg/modifier"
+	"github.com/awslabs/volume-modifier-for-k8s/pkg/util"
+	"github.com/kubernetes-csi/csi-lib-utils/leaderelection"
+	"github.com/kubernetes-csi/csi-lib-utils/metrics"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/informers"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/clientcmd"
+	"k8s.io/client-go/util/workqueue"
+	"k8s.io/klog/v2"
+)
+
+var (
+	clientConfigUrl = flag.String("client-config-url", "", "URL to build a client config from. Either this or kubeconfig needs to be set if the provisioner is being run out of cluster.")
+	kubeConfig      = flag.String("kubeconfig", "", "Absolute path to the kubeconfig")
+	resyncPeriod    = flag.Duration("resync-period", time.Minute*10, "Resync period for cache")
+	workers         = flag.Int("workers", 10, "Concurrency to process multiple modification requests")
+
+	csiAddress = flag.String("csi-address", "/run/csi/socket", "Address of the CSI driver socket.")
+	timeout    = flag.Duration("timeout", 10*time.Second, "Timeout for waiting for CSI driver socket.")
+
+	showVersion = flag.Bool("version", false, "Show version")
+
+	retryIntervalStart = flag.Duration("retry-interval-start", time.Second, "Initial retry interval of failed volume modification. It exponentially increases with each failure, up to retry-interval-max.")
+	retryIntervalMax   = flag.Duration("retry-interval-max", 5*time.Minute, "Maximum retry interval of failed volume modification.")
+
+	enableLeaderElection        = flag.Bool("leader-election", false, "Enable leader election.")
+	leaderElectionNamespace     = flag.String("leader-election-namespace", "", "Namespace where the leader election resource lives. Defaults to the pod namespace if not set.")
+	leaderElectionLeaseDuration = flag.Duration("leader-election-lease-duration", 15*time.Second, "Duration, in seconds, that non-leader candidates will wait to force acquire leadership. Defaults to 15 seconds.")
+	leaderElectionRenewDeadline = flag.Duration("leader-election-renew-deadline", 10*time.Second, "Duration, in seconds, that the acting leader will retry refreshing leadership before giving up. Defaults to 10 seconds.")
+	leaderElectionRetryPeriod   = flag.Duration("leader-election-retry-period", 5*time.Second, "Duration, in seconds, the LeaderElector clients should wait between tries of actions. Defaults to 5 seconds.")
+
+	httpEndpoint = flag.String("http-endpoint", "", "The TCP network address where the HTTP server for diagnostics, including metrics and leader election health check, will listen (example: `:8080`). The default is empty string, which means the server is disabled. Only one of `--metrics-address` and `--http-endpoint` can be set.")
+	metricsPath  = flag.String("metrics-path", "/metrics", "The HTTP path where prometheus metrics will be exposed. Default is `/metrics`.")
+
+	kubeAPIQPS   = flag.Float64("kube-api-qps", 5, "QPS to use while communicating with the kubernetes apiserver. Defaults to 5.0.")
+	kubeAPIBurst = flag.Int("kube-api-burst", 10, "Burst to use while communicating with the kubernetes apiserver. Defaults to 10.")
+
+	// Passed through ldflags.
+	version = "<unknown>"
+)
+
+func main() {
+	klog.InitFlags(nil)
+	flag.Set("logtostderr", "true")
+	flag.Parse()
+
+	if *showVersion {
+		fmt.Println(os.Args[0], version)
+		os.Exit(0)
+	}
+	klog.Infof("Version : %s", version)
+
+	addr := *httpEndpoint
+	var config *rest.Config
+	var err error
+	if *clientConfigUrl != "" || *kubeConfig != "" {
+		config, err = clientcmd.BuildConfigFromFlags(*clientConfigUrl, *kubeConfig)
+	} else {
+		config, err = rest.InClusterConfig()
+	}
+	if err != nil {
+		klog.Fatal(err.Error())
+	}
+
+	config.QPS = float32(*kubeAPIQPS)
+	config.Burst = *kubeAPIBurst
+
+	kubeClient, err := kubernetes.NewForConfig(config)
+	if err != nil {
+		klog.Fatal(err.Error())
+	}
+
+	informerFactory := informers.NewSharedInformerFactory(kubeClient, *resyncPeriod)
+	mux := http.NewServeMux()
+	metricsManager := metrics.NewCSIMetricsManager("" /* driverName */)
+	csiClient, err := csi.New(*csiAddress, *timeout, metricsManager)
+	if err != nil {
+		klog.Fatal(err.Error())
+	}
+	if err := csiClient.SupportsVolumeModification(context.TODO()); err != nil {
+		klog.Fatalf("CSI driver does not support volume modification: %v", err)
+	}
+
+	driverName, err := getDriverName(csiClient, *timeout)
+	if err != nil {
+		klog.Fatal(fmt.Errorf("get driver name failed: %v", err))
+	}
+	klog.V(2).Infof("CSI driver name: %q", driverName)
+
+	csiModifier, err := modifier.NewFromClient(
+		driverName,
+		csiClient,
+		kubeClient,
+		*timeout,
+	)
+	if err != nil {
+		klog.Fatal(err.Error())
+	}
+
+	if addr != "" {
+		metricsManager.RegisterToServer(mux, *metricsPath)
+		metricsManager.SetDriverName(driverName)
+		go func() {
+			klog.Infof("ServeMux listening at %q", addr)
+			err := http.ListenAndServe(addr, mux)
+			if err != nil {
+				klog.Fatalf("Failed to start HTTP server at specified address (%q) and metrics path (%q): %s", addr, *metricsPath, err)
+			}
+		}()
+	}
+
+	modifierName := csiModifier.Name()
+	mc := controller.NewModifyController(
+		modifierName,
+		csiModifier,
+		kubeClient,
+		*resyncPeriod,
+		informerFactory,
+		workqueue.NewItemExponentialFailureRateLimiter(*retryIntervalStart, *retryIntervalMax),
+		true, /* retryFailure */
+	)
+
+	run := func(ctx context.Context) {
+		informerFactory.Start(wait.NeverStop)
+		mc.Run(*workers, ctx)
+	}
+
+	if !*enableLeaderElection {
+		run(context.TODO())
+	} else {
+		lockName := "volume-modifier-for-k8s-" + util.SanitizeName(modifierName)
+		leKubeClient, err := kubernetes.NewForConfig(config)
+		if err != nil {
+			klog.Fatal(err.Error())
+		}
+		le := leaderelection.NewLeaderElection(leKubeClient, lockName, run)
+		if *httpEndpoint != "" {
+			le.PrepareHealthCheck(mux, leaderelection.DefaultHealthCheckTimeout)
+		}
+
+		if *leaderElectionNamespace != "" {
+			le.WithNamespace(*leaderElectionNamespace)
+		}
+
+		le.WithLeaseDuration(*leaderElectionLeaseDuration)
+		le.WithRenewDeadline(*leaderElectionRenewDeadline)
+		le.WithRetryPeriod(*leaderElectionRetryPeriod)
+
+		if err := le.Run(); err != nil {
+			klog.Fatalf("error initializing leader election: %v", err)
+		}
+	}
+}
+
+func getDriverName(client csi.Client, timeout time.Duration) (string, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	defer cancel()
+	return client.GetDriverName(ctx)
+}

go.mod

@@ -0,0 +1,67 @@
+module github.com/awslabs/volume-modifier-for-k8s
+
+go 1.19
+
+require (
+	github.com/golang/protobuf v1.5.3
+	github.com/google/go-cmp v0.5.9
+	github.com/kubernetes-csi/csi-lib-utils v0.13.0
+	google.golang.org/grpc v1.54.0
+	google.golang.org/protobuf v1.30.0
+	k8s.io/api v0.27.1
+	k8s.io/apimachinery v0.27.1
+	k8s.io/client-go v0.27.1
+	k8s.io/csi-translation-lib v0.27.1
+	k8s.io/klog/v2 v2.90.1
+	k8s.io/kubectl v0.27.1
+)
+
+require (
+	github.com/beorn7/perks v1.0.1 // indirect
+	github.com/blang/semver/v4 v4.0.0 // indirect
+	github.com/cespare/xxhash/v2 v2.2.0 // indirect
+	github.com/container-storage-interface/spec v1.7.0 // indirect
+	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/emicklei/go-restful/v3 v3.9.0 // indirect
+	github.com/evanphx/json-patch v4.12.0+incompatible // indirect
+	github.com/go-logr/logr v1.2.3 // indirect
+	github.com/go-openapi/jsonpointer v0.19.6 // indirect
+	github.com/go-openapi/jsonreference v0.20.1 // indirect
+	github.com/go-openapi/swag v0.22.3 // indirect
+	github.com/gogo/protobuf v1.3.2 // indirect
+	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
+	github.com/google/gnostic v0.5.7-v3refs // indirect
+	github.com/google/gofuzz v1.2.0 // indirect
+	github.com/google/uuid v1.3.0 // indirect
+	github.com/imdario/mergo v0.3.12 // indirect
+	github.com/josharian/intern v1.0.0 // indirect
+	github.com/json-iterator/go v1.1.12 // indirect
+	github.com/mailru/easyjson v0.7.7 // indirect
+	github.com/matttproud/golang_protobuf_extensions v1.0.2 // indirect
+	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
+	github.com/modern-go/reflect2 v1.0.2 // indirect
+	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
+	github.com/pkg/errors v0.9.1 // indirect
+	github.com/prometheus/client_golang v1.14.0 // indirect
+	github.com/prometheus/client_model v0.3.0 // indirect
+	github.com/prometheus/common v0.37.0 // indirect
+	github.com/prometheus/procfs v0.8.0 // indirect
+	github.com/spf13/pflag v1.0.5 // indirect
+	golang.org/x/net v0.8.0 // indirect
+	golang.org/x/oauth2 v0.4.0 // indirect
+	golang.org/x/sys v0.6.0 // indirect
+	golang.org/x/term v0.6.0 // indirect
+	golang.org/x/text v0.8.0 // indirect
+	golang.org/x/time v0.0.0-20220210224613-90d013bbcef8 // indirect
+	google.golang.org/appengine v1.6.7 // indirect
+	google.golang.org/genproto v0.0.0-20230110181048-76db0878b65f // indirect
+	gopkg.in/inf.v0 v0.9.1 // indirect
+	gopkg.in/yaml.v2 v2.4.0 // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
+	k8s.io/component-base v0.27.1 // indirect
+	k8s.io/kube-openapi v0.0.0-20230308215209-15aac26d736a // indirect
+	k8s.io/utils v0.0.0-20230209194617-a36077c30491 // indirect
+	sigs.k8s.io/json v0.0.0-20221116044647-bc3834ca7abd // indirect
+	sigs.k8s.io/structured-merge-diff/v4 v4.2.3 // indirect
+	sigs.k8s.io/yaml v1.3.0 // indirect
+)