udev-manager: new disk manager plugin based on udev

.gitignore

@@ -0,0 +1 @@
+.vscode

Dockerfile

@@ -1,14 +1,16 @@
-FROM golang:1.18 as build
+FROM golang:1.24-bookworm as build
 
 WORKDIR /go/app
 COPY . /go/app
 
-RUN CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -o ydb-disk-manager cmd/ydb-disk-manager/main.go
+RUN apt update && apt install -y libudev-dev
+RUN CGO_ENABLED=1 GOOS=linux GOARCH=amd64 go build -o ydb-disk-manager cmd/ydb-disk-manager/main.go
 
-FROM ubuntu:20.04
+FROM debian:bookworm
 
 WORKDIR /root
 
+RUN apt update && apt install -y libudev1 && apt clean
 COPY --from=build /go/app/ydb-disk-manager /usr/bin/ydb-disk-manager
 
 ENTRYPOINT ["/usr/bin/ydb-disk-manager"]

cmd/udev-manager/main.go

@@ -0,0 +1,279 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"flag"
+	"fmt"
+	"io"
+	"net/http"
+	"os"
+	"os/signal"
+	"regexp"
+	"strings"
+	"sync"
+	"syscall"
+
+	"gopkg.in/yaml.v3"
+
+	"k8s.io/klog/v2"
+
+	"github.com/ydb-platform/ydb-disk-manager/internal/mux"
+	"github.com/ydb-platform/ydb-disk-manager/internal/plugin"
+	"github.com/ydb-platform/ydb-disk-manager/internal/udev"
+)
+
+func main() {
+	appContext, appCancel := context.WithCancel(context.Background())
+	appWaitGroup := &sync.WaitGroup{}
+	defer appWaitGroup.Wait()
+	defer appCancel()
+
+	flags := initFlags()
+
+	// Registry creates a separate plugin for each resource registered
+	registry, err := plugin.NewRegistry(appContext, appWaitGroup)
+	if err != nil {
+		klog.Fatalf("failed to create plugin registry: %v", err)
+		os.Exit(1)
+	}
+
+	// udev discovery looks up devices and listens for system events
+	devDiscovery, err := udev.NewDiscovery(appWaitGroup)
+	if err != nil {
+		klog.Fatalf("failed to start udev discovery: %v", err)
+		os.Exit(1)
+	}
+	defer devDiscovery.Close()
+
+	domain := flags.config.DeviceDomain
+
+	var cancel mux.CancelFunc = mux.CancelFunc(appCancel)
+	for _, partConfig := range flags.config.Partitions {
+		cancel = mux.ChainCancelFunc(
+			plugin.NewScatter(
+				devDiscovery,
+				registry,
+				domain,
+				"part-",
+				plugin.PartitionLabelMatcher(domain, partConfig.matcher),
+			),
+			cancel,
+		)
+	}
+
+	klog.Info("Starting /healthz server on port :8080")
+	go func() {
+		http.HandleFunc("/healthz", registry.Healthz)
+		if err := http.ListenAndServe(":8080", nil); err != nil {
+			klog.Fatalf("failed to start /healthz server: %v", err)
+		}
+	}()
+
+	sigs := make(chan os.Signal, 1)
+	signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM)
+	defer cancel()
+	for signal := range sigs {
+		switch signal {
+		case syscall.SIGINT, syscall.SIGTERM:
+			klog.Infof("Received signal %q, shutting down", signal.String())
+			return
+		}
+	}
+}
+
+type configSource interface {
+	String() string
+	open() (io.Reader, func() error, error)
+}
+
+type fileConfigSource struct {
+	path string
+}
+
+func (fcs *fileConfigSource) open() (io.Reader, func() error, error) {
+	file, err := os.Open(fcs.path)
+	if err != nil {
+		return nil, nil, err
+	}
+	return file, file.Close, nil
+}
+
+func (fcs *fileConfigSource) String() string {
+	return "file:" + fcs.path
+}
+
+type envConfigSource struct {
+	variable string
+}
+
+func (ecs *envConfigSource) open() (io.Reader, func() error, error) {
+	data := os.Getenv(ecs.variable)
+	if data == "" {
+		return nil, nil, fmt.Errorf("config: environment variable %s is not set", ecs.variable)
+	}
+	return strings.NewReader(data), func() error { return nil }, nil
+}
+
+func (ecs *envConfigSource) String() string {
+	return "env:" + ecs.variable
+}
+
+type stdinConfigSource struct{}
+
+func (scs *stdinConfigSource) open() (io.Reader, func() error, error) {
+	return os.Stdin, func() error { return nil }, nil
+}
+
+func (scs *stdinConfigSource) String() string {
+	return "stdin"
+}
+
+type ConfigFlag struct {
+	configSource
+}
+
+func (cf *ConfigFlag) Set(value string) error {
+	if strings.HasPrefix(value, "file:") {
+		cf.configSource = &fileConfigSource{path: strings.TrimPrefix(value, "file:")}
+	} else if strings.HasPrefix(value, "env:") {
+		cf.configSource = &envConfigSource{variable: strings.TrimPrefix(value, "env:")}
+	} else if strings.HasPrefix(value, "stdin") {
+		cf.configSource = &stdinConfigSource{}
+	} else {
+		return fmt.Errorf("invalid config source: %s", value)
+	}
+
+	return nil
+}
+
+func (cf *ConfigFlag) String() string {
+	if cf.configSource == nil {
+		return ""
+	}
+	return cf.configSource.String()
+}
+
+type FlagValues struct {
+	Config ConfigFlag
+
+	config *Config
+}
+
+func initFlags() FlagValues {
+	values := FlagValues{}
+	flags := flag.NewFlagSet("udev-manager", flag.ExitOnError)
+	klog.InitFlags(flags)
+	flags.Var(&values.Config, "config", `configuration source (in form "file:<path>", "env:<ENV_VARIABLE>" or "stdin")`)
+	flags.Parse(os.Args[1:])
+	if values.Config.configSource == nil {
+		flags.Output().Write([]byte("config flag is required\n"))
+		flags.Usage()
+		os.Exit(2)
+	}
+	configReader, configCloser, err := values.Config.open()
+	if err != nil {
+		klog.Fatalf("failed to open --config %q: %v", values.Config.String(), err)
+		os.Exit(1)
+	}
+	defer configCloser()
+
+	config, err := parseConfig(configReader)
+	if err != nil {
+		klog.Fatalf("failed to parse --config %q: %v", values.Config.String(), err)
+		os.Exit(1)
+	}
+
+	values.config = config
+
+	return values
+}
+
+type PartitionsConfig struct {
+	Matcher string `yaml:"matcher"` // matcher should be a valid regular expression
+
+	matcher *regexp.Regexp // compiled matcher if the config is valid
+}
+
+func (pc *PartitionsConfig) validate() error {
+	// Compile the regular expression
+	matcher, err := regexp.Compile(pc.Matcher)
+	if err != nil {
+		return fmt.Errorf(".matcher: %q must be a valid regexp: %w", pc.Matcher, err)
+	}
+	if matcher.NumSubexp() > 1 {
+		return fmt.Errorf(".matcher: %q must have at most one capturing group", pc.Matcher)
+	}
+	pc.matcher = matcher
+	return nil
+}
+
+type HostDevConfig struct {
+	Matcher string `yaml:"matcher"` // matcher should be a valid regular expression
+	Prefix string `yaml:"prefix"`
+
+	matcher *regexp.Regexp // compiled matcher if the config is valid
+}
+
+func (hc *HostDevConfig) validate() error {
+	// Compile the regular expression
+	matcher, err := regexp.Compile(hc.Matcher)
+	if err != nil {
+		return fmt.Errorf(".matcher: %q must be a valid regexp: %w", hc.Matcher, err)
+	}
+	hc.matcher = matcher
+	return nil
+}
+
+type Config struct {
+	DeviceDomain string `yaml:"domain"`
+	Partitions []PartitionsConfig `yaml:"partitions"`
+	HostDevs   []HostDevConfig `yaml:"hostdevs"`
+}
+
+var (
+	deviceDomainRegex = regexp.MustCompile(`^([a-zA-Z0-9]([a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?\.)+[a-zA-Z0-9]([a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?$`)
+)
+
+func (c *Config) validate() error {
+	var errs error
+	// Validate the device domain
+	if c.DeviceDomain == "" {
+		errs = errors.Join(errs, fmt.Errorf(".domain: must be set"))
+	}
+	if !deviceDomainRegex.MatchString(c.DeviceDomain) {
+		errs = errors.Join(errs, fmt.Errorf(".domain: %q must be a valid domain name", c.DeviceDomain))
+	}
+
+	// Validate partitions
+	for i := range c.Partitions {
+		if err := c.Partitions[i].validate(); err != nil {
+			errs = errors.Join(errs, fmt.Errorf(".partitions[%d]%w", i, err))
+		}
+	}
+
+	// Validate hostdevs
+	for i := range c.HostDevs {
+		if err := c.HostDevs[i].validate(); err != nil {
+			errs = errors.Join(errs, fmt.Errorf(".hostdevs[%d]%w", i, err))
+		}
+	}
+
+	return nil
+}
+
+func parseConfig(reader io.Reader) (*Config, error) {
+	// Parse the config file
+	decoder := yaml.NewDecoder(reader)
+	config := &Config{}
+	if err := decoder.Decode(config); err != nil {
+		return nil, err
+	}
+
+	// Validate the config
+	if err := config.validate(); err != nil {
+		return nil, err
+	}
+
+	return config, nil
+}

cmd/ydb-disk-manager/main.go

@@ -3,7 +3,6 @@ package main
 import (
 	"flag"
 	"fmt"
-	"io/ioutil"
 	"net/http"
 	"os"
 	"os/signal"
@@ -45,7 +44,7 @@ func main() {
 
 	// Setting up the disks to check
 	klog.V(0).Infof("Reading configuration file %s", confFileName)
-	yamlFile, err := ioutil.ReadFile(confFileName)
+	yamlFile, err := os.ReadFile(confFileName)
 	if err != nil {
 		klog.Fatalf("Reading configuration file failed with: %s", err)
 	}

go.mod

@@ -1,25 +1,37 @@
 module github.com/ydb-platform/ydb-disk-manager
 
-go 1.18
+go 1.23.0
+
+toolchain go1.24.1
 
 require (
 	github.com/fsnotify/fsnotify v1.6.0
-	golang.org/x/net v0.12.0
+	github.com/jochenvg/go-udev v0.0.0-20240801134859-b65ed646224b
+	github.com/kennygrant/sanitize v1.2.4
+	github.com/onsi/ginkgo/v2 v2.23.0
+	github.com/onsi/gomega v1.36.2
+	golang.org/x/net v0.36.0
 	google.golang.org/grpc v1.56.2
-	google.golang.org/protobuf v1.31.0
+	google.golang.org/protobuf v1.36.1
 	gopkg.in/yaml.v2 v2.4.0
+	gopkg.in/yaml.v3 v3.0.1
 	k8s.io/klog/v2 v2.70.1
 	k8s.io/kubelet v0.25.3
 )
 
 require (
-	github.com/go-logr/logr v1.2.3 // indirect
+	github.com/go-logr/logr v1.4.2 // indirect
+	github.com/go-task/slim-sprig/v3 v3.0.0 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang/protobuf v1.5.3 // indirect
+	github.com/google/go-cmp v0.6.0 // indirect
+	github.com/google/pprof v0.0.0-20241210010833-40e02aabc2ad // indirect
+	github.com/jkeiser/iter v0.0.0-20200628201005-c8aa0ae784d1 // indirect
 	github.com/kr/pretty v0.3.1 // indirect
 	github.com/rogpeppe/go-internal v1.11.0 // indirect
-	golang.org/x/sys v0.10.0 // indirect
-	golang.org/x/text v0.11.0 // indirect
+	golang.org/x/sys v0.30.0 // indirect
+	golang.org/x/text v0.22.0 // indirect
+	golang.org/x/tools v0.30.0 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20230724170836-66ad5b6ff146 // indirect
 	gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c // indirect
 )