diff --git a/go.mod b/go.mod index e7120a5..e2dcc14 100644 --- a/go.mod +++ b/go.mod @@ -17,8 +17,8 @@ require ( go.fd.io/govpp/extras v0.1.1-0.20261019084134-c02eb51f8588 golang.org/x/net v0.61.0 golang.org/x/sys v0.49.0 - google.golang.org/genproto/googleapis/api v0.0.0-20261026165753-2ffab32bce7e - google.golang.org/grpc v1.84.0 + google.golang.org/genproto/googleapis/api v0.0.0-20261104224025-d1e00e9f33ab + google.golang.org/grpc v1.85.0 google.golang.org/protobuf v1.36.12 modernc.org/sqlite v1.61.0 ) @@ -27,6 +27,7 @@ require ( github.com/miekg/dns v1.1.73 github.com/onsi/ginkgo/v2 v2.34.0 github.com/onsi/gomega v1.44.0 + github.com/voidstarit/kouja v0.0.0-20261107022657-5dd27b7c76f0 ) require ( @@ -58,7 +59,7 @@ require ( github.com/vmihailenco/tagparser/v2 v2.0.0 // indirect golang.org/x/mod v0.42.0 // indirect golang.org/x/sync v0.24.0 // indirect - golang.org/x/time v0.7.0 // indirect + golang.org/x/time v0.15.0 // indirect golang.org/x/tools v0.52.0 // indirect ) @@ -92,7 +93,7 @@ require ( golang.org/x/crypto v0.58.0 // indirect golang.org/x/term v0.47.0 // indirect golang.org/x/text v0.43.0 // indirect - google.golang.org/genproto/googleapis/rpc v0.0.0-20261019180522-ce908d29299b // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20261019180522-ce908d29299b gvisor.dev/gvisor v0.0.0-20250503011706-39ed1f5ac29c modernc.org/libc v1.79.0 // indirect modernc.org/mathutil v1.7.1 // indirect diff --git a/go.sum b/go.sum index 05bd370..45a7156 100644 --- a/go.sum +++ b/go.sum @@ -189,6 +189,8 @@ github.com/vmihailenco/msgpack/v5 v5.4.1 h1:cQriyiUvjTwOHg8QZaPihLWeRAAVoCpE00IU github.com/vmihailenco/msgpack/v5 v5.4.1/go.mod h1:GaZTsDaehaPpQVyxrf5mtQlH+pc21PIudVV/E3rRQok= github.com/vmihailenco/tagparser/v2 v2.0.0 h1:y09buUbR+b5aycVFQs/g70pqKVZNBmxwAhO7/IwNM9g= github.com/vmihailenco/tagparser/v2 v2.0.0/go.mod h1:Wri+At7QHww0WTrCBeu4J6bNtoV6mEfg5OIWRZA9qds= +github.com/voidstarit/kouja v0.0.0-20261107022657-5dd27b7c76f0 h1:+HjRG4ohgOVq3+lfXC1tVUtAW4bT8C+AbtapkIoHBjY= +github.com/voidstarit/kouja v0.0.0-20261107022657-5dd27b7c76f0/go.mod h1:SjdnoRUyB7vwAmvs3fVU/XIPzqYdt2hQ0C4RULcftxM= github.com/x-cray/logrus-prefixed-formatter v0.5.2 h1:00txxvfBM9muc0jiLIEAkAcIMJzfthRT6usrui8uGmg= github.com/x-cray/logrus-prefixed-formatter v0.5.2/go.mod h1:2duySbKsL6M18s5GU7VPsoEPHyzalCE06qoARUCeBBE= go.fd.io/govpp v0.13.0 h1:MnjH9I5K+X0860CeeuBcMSu3uyUKA6X9AenKzdiGpnA= @@ -230,20 +232,20 @@ golang.org/x/term v0.47.0/go.mod h1:efAfAuG6RKABvg1ArGebRkvX6LXul14SM67o2e4mUCY= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.43.0 h1:1QivrlhwAsnMOcqgOdl9my7s9OyZl8O5/UGUTJVXoWc= golang.org/x/text v0.43.0/go.mod h1:hbSIYA/amXcRXTFSZqh8tWHEoqRvvbKPJ2uz/KU+a+s= -golang.org/x/time v0.7.0 h1:ntUhktv3OPE6TgYxXWv9vKvUSJyIFJlyohwbkEwPrKQ= -golang.org/x/time v0.7.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM= +golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U= +golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno= golang.org/x/tools v0.0.0-20200130002326-2f3ba24bd6e7/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28= golang.org/x/tools v0.52.0 h1:ryxg4U63oaUYk7uOLTF37n43f4qgtGvqV4Ioit1BaEs= golang.org/x/tools v0.52.0/go.mod h1:ItcP5UNaVGG+Yc4eIYmaFLNJ4soEopkx5Y2d4lcEurs= golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4= gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E= -google.golang.org/genproto/googleapis/api v0.0.0-20261026165753-2ffab32bce7e h1:e789sLZjeTdc02ZCQGTSzHOu0fR73+dwPQVKUp67Z88= -google.golang.org/genproto/googleapis/api v0.0.0-20261026165753-2ffab32bce7e/go.mod h1:vHV3wJM9WZyfV7GyB+OQGq5dWVrFwCDxapPK4WlRD88= +google.golang.org/genproto/googleapis/api v0.0.0-20261104224025-d1e00e9f33ab h1:n6IDWZ1RzI/NberLoLLFNrjl1H45uhc4+WYsUf5lPPc= +google.golang.org/genproto/googleapis/api v0.0.0-20261104224025-d1e00e9f33ab/go.mod h1:vHV3wJM9WZyfV7GyB+OQGq5dWVrFwCDxapPK4WlRD88= google.golang.org/genproto/googleapis/rpc v0.0.0-20261019180522-ce908d29299b h1:6WNWHZk91dp4gZ25642OALkuG1UVNZbwENJD19DnB8k= google.golang.org/genproto/googleapis/rpc v0.0.0-20261019180522-ce908d29299b/go.mod h1:OaIUM3+LpYcK2GXM4FTmhWoIq371Owdr+Cc7/BsYHHc= -google.golang.org/grpc v1.84.0 h1:soMyaPJ8pAak5PIQ0DGBUir0XRo2fRoMqhNWMLlLxO0= -google.golang.org/grpc v1.84.0/go.mod h1:ljCht0DrxQrXBDRTZp52Qxh3Ffk8CdYm2sj4O2QN2C0= +google.golang.org/grpc v1.85.0 h1:BPBCfuVRQ1qvBhKDOq9uvQllMvO59rKsYhXpp7Dd8vw= +google.golang.org/grpc v1.85.0/go.mod h1:0rQrOD58yjal/0EaVbGBANindixsCZNQPDV3VKtaKuQ= google.golang.org/protobuf v1.36.12 h1:pJOKDDOyeXErUroCihFAd5LQuwXBSpVnKGrj5o/fwxc= google.golang.org/protobuf v1.36.12/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= diff --git a/internal/kouja/client.go b/internal/kouja/client.go new file mode 100644 index 0000000..f31e154 --- /dev/null +++ b/internal/kouja/client.go @@ -0,0 +1,76 @@ +package kouja + +import ( + "context" + + koujacli "github.com/voidstarit/kouja/gen/client" + koujav1 "github.com/voidstarit/kouja/gen/kouja/v1" + "google.golang.org/grpc" + + "github.com/voidstarit/rookery/internal/store" +) + +// API is the subset of the kouja VolumeService used by Rookery. +type API interface { + CreateVolume(context.Context, *koujav1.CreateVolumeRequest, ...grpc.CallOption) (*koujav1.CreateVolumeResponse, error) + GetVolume(context.Context, *koujav1.GetVolumeRequest, ...grpc.CallOption) (*koujav1.GetVolumeResponse, error) + GetCredential(context.Context, *koujav1.GetCredentialRequest, ...grpc.CallOption) (*koujav1.GetCredentialResponse, error) + UnexportVolume(context.Context, *koujav1.UnexportVolumeRequest, ...grpc.CallOption) (*koujav1.UnexportVolumeResponse, error) + DeleteVolume(context.Context, *koujav1.DeleteVolumeRequest, ...grpc.CallOption) (*koujav1.DeleteVolumeResponse, error) + GetInfo(context.Context, *koujav1.GetInfoRequest, ...grpc.CallOption) (*koujav1.GetInfoResponse, error) + ListStorageClasses(context.Context, *koujav1.ListStorageClassesRequest, ...grpc.CallOption) (*koujav1.ListStorageClassesResponse, error) +} + +// Client is a kouja VolumeService client with an owned connection. +type Client struct { + API + close func() error +} + +// NewClient creates a kouja mTLS client for cfg. +func NewClient(cfg Config) (*Client, error) { + tlsConfig, err := koujacli.LoadTLSConfig(cfg.ClientCertFile, cfg.ClientKeyFile, cfg.CAFile) + if err != nil { + return nil, err + } + if cfg.ServerName != "" { + tlsConfig.ServerName = cfg.ServerName + } + grpcClient, err := koujacli.NewGRPC(cfg.Endpoint, tlsConfig) + if err != nil { + return nil, err + } + return &Client{API: grpcClient, close: grpcClient.Close}, nil +} + +// NewClientFromService wraps an existing service client. It is intended for +// tests and for callers that own the gRPC connection. +func NewClientFromService(service API) *Client { + return &Client{API: service} +} + +// Close closes the owned gRPC connection, if any. +func (c *Client) Close() error { + if c == nil || c.close == nil { + return nil + } + return c.close() +} + +// Factory builds a client per backend. It never shares a connection between +// backends with different endpoints or certificate identities. +type Factory struct{} + +// New constructs a client for a backend. +func (Factory) New(backend store.StorageBackend) (*Client, error) { + cfg, err := ParseBackend(backend) + if err != nil { + return nil, err + } + return NewClient(cfg) +} + +// NewClientForBackend is a convenience wrapper around Factory.New. +func NewClientForBackend(backend store.StorageBackend) (*Client, error) { + return Factory{}.New(backend) +} diff --git a/internal/kouja/client_test.go b/internal/kouja/client_test.go new file mode 100644 index 0000000..8a9401c --- /dev/null +++ b/internal/kouja/client_test.go @@ -0,0 +1,65 @@ +package kouja + +import ( + "testing" + + "github.com/voidstarit/rookery/internal/store" +) + +func TestNewClientAndFactory(t *testing.T) { + cfg, err := ParseConfig(testKoujaConfig(t)) + if err != nil { + t.Fatal(err) + } + client, err := NewClient(cfg) + if err != nil { + t.Fatalf("NewClient() = %v", err) + } + if client.API == nil { + t.Fatal("NewClient() returned a nil API") + } + if err := client.Close(); err != nil { + t.Fatalf("Close() = %v", err) + } + + backend := store.StorageBackend{Config: testKoujaConfig(t)} + factoryClient, err := Factory{}.New(backend) + if err != nil { + t.Fatalf("Factory.New() = %v", err) + } + if err := factoryClient.Close(); err != nil { + t.Fatal(err) + } + convenience, err := NewClientForBackend(backend) + if err != nil { + t.Fatal(err) + } + if err := convenience.Close(); err != nil { + t.Fatal(err) + } +} + +func TestNewClientFromServiceAndNilClose(t *testing.T) { + client := NewClientFromService(&fakeAPI{}) + if client == nil || client.API == nil { + t.Fatal("NewClientFromService() returned nil") + } + if err := client.Close(); err != nil { + t.Fatal(err) + } + var nilClient *Client + if err := nilClient.Close(); err != nil { + t.Fatal(err) + } +} + +func TestNewClientInvalidTLS(t *testing.T) { + cfg := Config{ + Endpoint: "10.0.0.1:9443", + CAFile: "/does/not/exist/ca.crt", ClientCertFile: "/does/not/exist/client.crt", + ClientKeyFile: "/does/not/exist/client.key", + } + if _, err := NewClient(cfg); err == nil { + t.Fatal("NewClient() accepted missing TLS files") + } +} diff --git a/internal/kouja/config.go b/internal/kouja/config.go new file mode 100644 index 0000000..5a69f7f --- /dev/null +++ b/internal/kouja/config.go @@ -0,0 +1,196 @@ +package kouja + +import ( + "crypto/tls" + "crypto/x509" + "errors" + "fmt" + "net" + "net/url" + "os" + "path/filepath" + "strconv" + "strings" + + "github.com/voidstarit/rookery/internal/blockdev" + "github.com/voidstarit/rookery/internal/store" +) + +var ( + // ErrNotConfigured is returned when a backend has no kouja configuration. + ErrNotConfigured = errors.New("kouja backend configuration is missing") + // ErrUnsupportedProtocol is returned for a protocol this build cannot use. + ErrUnsupportedProtocol = errors.New("unsupported kouja protocol") +) + +// Config is the validated kouja portion of a storage backend. +type Config struct { + Endpoint string + Protocol blockdev.Protocol + StorageClass string + Initiator string + HostNQN string + CAFile string + ClientCertFile string + ClientKeyFile string + ServerName string + Portal string + Owner string +} + +// ParseConfig validates a stored kouja backend config. +func ParseConfig(cfg store.StorageBackendConfig) (Config, error) { + if cfg.Kouja == nil { + return Config{}, ErrNotConfigured + } + raw := cfg.Kouja + protocol := blockdev.Protocol(strings.ToLower(strings.TrimSpace(raw.Protocol))) + if protocol == "" { + protocol = blockdev.ProtocolISCSI + } + if protocol != blockdev.ProtocolISCSI && protocol != blockdev.ProtocolNVMeTCP { + return Config{}, fmt.Errorf("%w: %q", ErrUnsupportedProtocol, raw.Protocol) + } + parsed := Config{ + Endpoint: strings.TrimSpace(raw.Endpoint), + Protocol: protocol, + StorageClass: strings.TrimSpace(raw.StorageClass), + Initiator: strings.TrimSpace(raw.Initiator), + HostNQN: strings.TrimSpace(raw.HostNQN), + CAFile: strings.TrimSpace(raw.CAFile), + ClientCertFile: strings.TrimSpace(raw.ClientCertFile), + ClientKeyFile: strings.TrimSpace(raw.ClientKeyFile), + ServerName: strings.TrimSpace(raw.ServerName), + Portal: strings.TrimSpace(raw.Portal), + } + if err := validateEndpoint(parsed.Endpoint); err != nil { + return Config{}, err + } + if parsed.StorageClass == "" { + return Config{}, errors.New("kouja storage_class is required") + } + if len(parsed.StorageClass) > 127 { + return Config{}, errors.New("kouja storage_class is too long") + } + for name, path := range map[string]string{ + "ca_file": parsed.CAFile, "client_cert_file": parsed.ClientCertFile, "client_key_file": parsed.ClientKeyFile, + } { + if path == "" || !filepath.IsAbs(path) { + return Config{}, fmt.Errorf("%s must be an absolute path", name) + } + } + if _, err := os.Stat(parsed.CAFile); err != nil { + return Config{}, fmt.Errorf("stat CA file: %w", err) + } + keyInfo, err := os.Stat(parsed.ClientKeyFile) + if err != nil { + return Config{}, fmt.Errorf("stat client key: %w", err) + } + if keyInfo.Mode().Perm()&0o004 != 0 { + return Config{}, errors.New("client_key_file must not be world-readable") + } + switch protocol { + case blockdev.ProtocolISCSI: + if !isIQN(parsed.Initiator) { + return Config{}, errors.New("kouja initiator must be an iSCSI IQN") + } + case blockdev.ProtocolNVMeTCP: + if !isNQN(parsed.HostNQN) { + return Config{}, errors.New("kouja host_nqn must be an NVMe NQN") + } + } + if parsed.Portal != "" { + if err := validatePortal(parsed.Portal); err != nil { + return Config{}, err + } + } + owner, err := ownerFromCertificate(parsed.ClientCertFile, parsed.ClientKeyFile) + if err != nil { + return Config{}, err + } + parsed.Owner = owner + return parsed, nil +} + +// ParseBackend validates the kouja config from a storage backend. +func ParseBackend(backend store.StorageBackend) (Config, error) { + return ParseConfig(backend.Config) +} + +func validateEndpoint(value string) error { + if value == "" { + return errors.New("kouja endpoint is required") + } + target := value + if strings.HasPrefix(value, "http://") { + return errors.New("kouja endpoint must use TLS (https://host:port or host:port)") + } + if strings.HasPrefix(value, "https://") { + parsed, err := url.Parse(value) + if err != nil { + return fmt.Errorf("parse kouja endpoint: %w", err) + } + if parsed.Host == "" || (parsed.Path != "" && parsed.Path != "/") { + return errors.New("kouja endpoint must be host:port or https://host:port") + } + target = parsed.Host + } + host, port, err := net.SplitHostPort(target) + if err != nil || strings.TrimSpace(host) == "" { + return errors.New("kouja endpoint must include host and port") + } + number, err := strconv.Atoi(port) + if err != nil || number < 1 || number > 65535 { + return errors.New("kouja endpoint port must be between 1 and 65535") + } + return nil +} + +func validatePortal(value string) error { + host, port, err := net.SplitHostPort(value) + if err != nil || strings.TrimSpace(host) == "" { + return errors.New("kouja portal must be host:port") + } + number, err := strconv.Atoi(port) + if err != nil || number < 1 || number > 65535 { + return errors.New("kouja portal port must be between 1 and 65535") + } + if ip := net.ParseIP(host); ip != nil && (ip.IsUnspecified() || ip.IsLoopback()) { + return errors.New("kouja portal must not be loopback or unspecified") + } + return nil +} + +func isIQN(value string) bool { + return strings.HasPrefix(strings.ToLower(strings.TrimSpace(value)), "iqn.") +} + +func isNQN(value string) bool { + return strings.HasPrefix(strings.ToLower(strings.TrimSpace(value)), "nqn.") +} + +func ownerFromCertificate(certFile, keyFile string) (string, error) { + pair, err := tls.LoadX509KeyPair(certFile, keyFile) + if err != nil { + return "", fmt.Errorf("load kouja client certificate: %w", err) + } + if len(pair.Certificate) == 0 { + return "", errors.New("kouja client certificate is empty") + } + cert, err := x509.ParseCertificate(pair.Certificate[0]) + if err != nil { + return "", fmt.Errorf("parse kouja client certificate: %w", err) + } + if len(cert.URIs) != 1 { + return "", errors.New("kouja client certificate must have exactly one URI SAN") + } + uri := cert.URIs[0] + if uri.Scheme != "spiffe" || uri.Host != "kouja" { + return "", errors.New("kouja client certificate URI SAN must use spiffe://kouja/caller/") + } + parts := strings.Split(strings.Trim(uri.Path, "/"), "/") + if len(parts) != 2 || parts[0] != "caller" || parts[1] == "" { + return "", errors.New("kouja client certificate URI SAN must use spiffe://kouja/caller/") + } + return uri.String(), nil +} diff --git a/internal/kouja/config_test.go b/internal/kouja/config_test.go new file mode 100644 index 0000000..4f228eb --- /dev/null +++ b/internal/kouja/config_test.go @@ -0,0 +1,207 @@ +package kouja + +import ( + "crypto/rand" + "crypto/rsa" + "crypto/x509" + "crypto/x509/pkix" + "encoding/pem" + "math/big" + "net/url" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/voidstarit/rookery/internal/store" +) + +func writeTestClientCert(t *testing.T, uris []*url.URL) (string, string, string) { + t.Helper() + key, err := rsa.GenerateKey(rand.Reader, 2048) + if err != nil { + t.Fatal(err) + } + template := &x509.Certificate{ + SerialNumber: big.NewInt(1), + Subject: pkix.Name{CommonName: "rookery"}, + NotBefore: time.Now().Add(-time.Hour), + NotAfter: time.Now().Add(time.Hour), + KeyUsage: x509.KeyUsageDigitalSignature | x509.KeyUsageKeyEncipherment | x509.KeyUsageCertSign, + ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageClientAuth, x509.ExtKeyUsageServerAuth}, + BasicConstraintsValid: true, + IsCA: true, + URIs: uris, + } + der, err := x509.CreateCertificate(rand.Reader, template, template, &key.PublicKey, key) + if err != nil { + t.Fatal(err) + } + dir := t.TempDir() + certFile := filepath.Join(dir, "client.crt") + keyFile := filepath.Join(dir, "client.key") + caFile := filepath.Join(dir, "ca.crt") + certPEM := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der}) + keyDER, err := x509.MarshalPKCS8PrivateKey(key) + if err != nil { + t.Fatal(err) + } + keyPEM := pem.EncodeToMemory(&pem.Block{Type: "PRIVATE KEY", Bytes: keyDER}) + if err := os.WriteFile(certFile, certPEM, 0o600); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(caFile, certPEM, 0o600); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(keyFile, keyPEM, 0o600); err != nil { + t.Fatal(err) + } + return certFile, keyFile, caFile +} + +func testKoujaConfig(t *testing.T) store.StorageBackendConfig { + t.Helper() + uri, err := url.Parse("spiffe://kouja/caller/rookery") + if err != nil { + t.Fatal(err) + } + certFile, keyFile, caFile := writeTestClientCert(t, []*url.URL{uri}) + return store.StorageBackendConfig{Kouja: &store.KoujaConfig{ + Endpoint: "10.0.0.1:9443", + StorageClass: "nvme", + Initiator: "iqn.2026-11.io.example:rookery", + HostNQN: "nqn.2026-11.io.example:rookery", + CAFile: caFile, + ClientCertFile: certFile, + ClientKeyFile: keyFile, + }} +} + +func TestParseConfig(t *testing.T) { + cfg, err := ParseConfig(testKoujaConfig(t)) + if err != nil { + t.Fatalf("ParseConfig() = %v", err) + } + if cfg.Protocol != "iscsi" || cfg.StorageClass != "nvme" || cfg.Owner != "spiffe://kouja/caller/rookery" { + t.Fatalf("parsed config = %+v", cfg) + } + if _, err := ParseBackend(store.StorageBackend{Config: testKoujaConfig(t)}); err != nil { + t.Fatalf("ParseBackend() = %v", err) + } +} + +func TestParseConfigNVMe(t *testing.T) { + cfg := testKoujaConfig(t) + cfg.Kouja.Protocol = "nvme-tcp" + parsed, err := ParseConfig(cfg) + if err != nil { + t.Fatalf("ParseConfig(nvme) = %v", err) + } + if parsed.Protocol != "nvme-tcp" || parsed.HostNQN == "" { + t.Fatalf("parsed nvme = %+v", parsed) + } +} + +func TestParseConfigValidationBranches(t *testing.T) { + base := testKoujaConfig(t) + tests := []struct { + name string + cfg store.StorageBackendConfig + want string + }{ + {name: "missing", cfg: store.StorageBackendConfig{}, want: "missing"}, + {name: "protocol", cfg: func() store.StorageBackendConfig { + value := base + value.Kouja = cloneKouja(base.Kouja) + value.Kouja.Protocol = "fc" + return value + }(), want: "unsupported"}, + {name: "endpoint", cfg: func() store.StorageBackendConfig { + value := base + value.Kouja = cloneKouja(base.Kouja) + value.Kouja.Endpoint = "http://host:9443" + return value + }(), want: "TLS"}, + {name: "endpoint port", cfg: func() store.StorageBackendConfig { + value := base + value.Kouja = cloneKouja(base.Kouja) + value.Kouja.Endpoint = "host" + return value + }(), want: "host and port"}, + {name: "class", cfg: func() store.StorageBackendConfig { + value := base + value.Kouja = cloneKouja(base.Kouja) + value.Kouja.StorageClass = "" + return value + }(), want: "storage_class"}, + {name: "relative ca", cfg: func() store.StorageBackendConfig { + value := base + value.Kouja = cloneKouja(base.Kouja) + value.Kouja.CAFile = "ca.crt" + return value + }(), want: "absolute"}, + {name: "world readable key", cfg: func() store.StorageBackendConfig { + value := testKoujaConfig(t) + if err := os.Chmod(value.Kouja.ClientKeyFile, 0o644); err != nil { + t.Fatal(err) + } + return value + }(), want: "world-readable"}, + {name: "initiator", cfg: func() store.StorageBackendConfig { + value := base + value.Kouja = cloneKouja(base.Kouja) + value.Kouja.Initiator = "host" + return value + }(), want: "IQN"}, + {name: "nvme host nqn", cfg: func() store.StorageBackendConfig { + value := base + value.Kouja = cloneKouja(base.Kouja) + value.Kouja.Protocol = "nvme-tcp" + value.Kouja.HostNQN = "host" + return value + }(), want: "NQN"}, + {name: "loopback portal", cfg: func() store.StorageBackendConfig { + value := base + value.Kouja = cloneKouja(base.Kouja) + value.Kouja.Portal = "127.0.0.1:3260" + return value + }(), want: "loopback"}, + {name: "unspecified portal", cfg: func() store.StorageBackendConfig { + value := base + value.Kouja = cloneKouja(base.Kouja) + value.Kouja.Portal = "0.0.0.0:3260" + return value + }(), want: "loopback or unspecified"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + _, err := ParseConfig(tt.cfg) + if err == nil || !strings.Contains(err.Error(), tt.want) { + t.Fatalf("ParseConfig() = %v, want %q", err, tt.want) + } + }) + } +} + +func TestOwnerFromCertificateValidation(t *testing.T) { + uri, _ := url.Parse("spiffe://kouja/caller/rookery") + certFile, keyFile, _ := writeTestClientCert(t, nil) + if _, err := ownerFromCertificate(certFile, keyFile); err == nil { + t.Fatal("ownerFromCertificate accepted no URI SAN") + } + certFile, keyFile, _ = writeTestClientCert(t, []*url.URL{uri, uri}) + if _, err := ownerFromCertificate(certFile, keyFile); err == nil { + t.Fatal("ownerFromCertificate accepted multiple URI SANs") + } + bad, _ := url.Parse("spiffe://other/caller/rookery") + certFile, keyFile, _ = writeTestClientCert(t, []*url.URL{bad}) + if _, err := ownerFromCertificate(certFile, keyFile); err == nil { + t.Fatal("ownerFromCertificate accepted wrong URI SAN") + } +} + +func cloneKouja(value *store.KoujaConfig) *store.KoujaConfig { + copy := *value + return © +} diff --git a/internal/kouja/errors.go b/internal/kouja/errors.go new file mode 100644 index 0000000..231263c --- /dev/null +++ b/internal/kouja/errors.go @@ -0,0 +1,90 @@ +package kouja + +import ( + "errors" + + koujacli "github.com/voidstarit/kouja/gen/client" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +// CleanupClass tells the deletion worker whether an error is terminal, +// retryable, or already represents the desired end state. +type CleanupClass int + +const ( + // CleanupTerminal means the operation should enter delete_failed. + CleanupTerminal CleanupClass = iota + // CleanupRetry means the operation should be retried with backoff. + CleanupRetry + // CleanupSuccess means the desired end state already exists. + CleanupSuccess +) + +// Reason returns the kouja ErrorInfo reason when present. +func Reason(err error) string { + if info, ok := koujacli.ErrorInfo(err); ok { + return info.GetReason() + } + return "" +} + +// IsNotFound reports whether an error means the resource is already gone. +func IsNotFound(err error) bool { + if err == nil { + return false + } + return Reason(err) == "NOT_FOUND" || status.Code(err) == codes.NotFound +} + +// IsSessionBlocked reports whether kouja refused a delete because a session +// state is active or unknown. Rookery retries these with backoff. +func IsSessionBlocked(err error) bool { + reason := Reason(err) + return reason == "SESSION_ACTIVE" || reason == "SESSION_UNKNOWN" +} + +// IsRetryable exposes the generated client's retry policy. +func IsRetryable(err error) bool { + return err != nil && koujacli.IsRetryable(err) +} + +// ClassifyCleanup maps an error to the deletion worker's policy. +func ClassifyCleanup(err error) CleanupClass { + if err == nil { + return CleanupSuccess + } + if IsNotFound(err) { + return CleanupSuccess + } + if IsSessionBlocked(err) || Reason(err) == "OPERATION_IN_PROGRESS" { + return CleanupRetry + } + if koujacli.IsRetryable(err) { + return CleanupRetry + } + return CleanupTerminal +} + +// ClassifyError returns Retry for the generated client's retryable errors and +// Terminal otherwise. It exists for callers that do not need cleanup-specific +// NOT_FOUND handling. +func ClassifyError(err error) CleanupClass { + if err == nil { + return CleanupSuccess + } + if koujacli.IsRetryable(err) { + return CleanupRetry + } + return CleanupTerminal +} + +// IsTerminal reports whether ClassifyCleanup returns CleanupTerminal. +func IsTerminal(err error) bool { + return ClassifyCleanup(err) == CleanupTerminal +} + +// IsUnsupported reports whether an error is an unsupported protocol. +func IsUnsupported(err error) bool { + return errors.Is(err, ErrUnsupportedProtocol) +} diff --git a/internal/kouja/errors_test.go b/internal/kouja/errors_test.go new file mode 100644 index 0000000..4fa6b27 --- /dev/null +++ b/internal/kouja/errors_test.go @@ -0,0 +1,75 @@ +package kouja + +import ( + "errors" + "testing" + + errdetails "google.golang.org/genproto/googleapis/rpc/errdetails" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +func errorWithReason(t *testing.T, code codes.Code, reason string) error { + t.Helper() + base := status.New(code, reason) + withDetails, err := base.WithDetails(&errdetails.ErrorInfo{Domain: "kouja.v1", Reason: reason}) + if err != nil { + t.Fatal(err) + } + return withDetails.Err() +} + +func TestClassifyCleanup(t *testing.T) { + if ClassifyCleanup(nil) != CleanupSuccess { + t.Fatal("nil error is not success") + } + if ClassifyCleanup(status.Error(codes.NotFound, "missing")) != CleanupSuccess { + t.Fatal("NotFound is not success") + } + if ClassifyCleanup(errorWithReason(t, codes.NotFound, "NOT_FOUND")) != CleanupSuccess { + t.Fatal("NOT_FOUND is not success") + } + for _, reason := range []string{"SESSION_ACTIVE", "SESSION_UNKNOWN", "OPERATION_IN_PROGRESS"} { + if ClassifyCleanup(errorWithReason(t, codes.FailedPrecondition, reason)) != CleanupRetry { + t.Fatalf("%s is not retry", reason) + } + } + if ClassifyCleanup(status.Error(codes.Unavailable, "down")) != CleanupRetry { + t.Fatal("Unavailable is not retry") + } + if ClassifyCleanup(status.Error(codes.InvalidArgument, "bad")) != CleanupTerminal { + t.Fatal("InvalidArgument is not terminal") + } + if ClassifyError(nil) != CleanupSuccess || ClassifyError(status.Error(codes.Unavailable, "down")) != CleanupRetry { + t.Fatal("ClassifyError mismatch") + } +} + +func TestErrorHelpers(t *testing.T) { + notFound := errorWithReason(t, codes.NotFound, "NOT_FOUND") + if !IsNotFound(notFound) || !IsNotFound(status.Error(codes.NotFound, "x")) { + t.Fatal("IsNotFound() = false") + } + if IsNotFound(errors.New("other")) { + t.Fatal("IsNotFound() accepted unrelated error") + } + if !IsSessionBlocked(errorWithReason(t, codes.FailedPrecondition, "SESSION_ACTIVE")) || + !IsSessionBlocked(errorWithReason(t, codes.FailedPrecondition, "SESSION_UNKNOWN")) { + t.Fatal("IsSessionBlocked() = false") + } + if Reason(notFound) != "NOT_FOUND" { + t.Fatalf("Reason() = %q", Reason(notFound)) + } + if Reason(errors.New("other")) != "" { + t.Fatal("Reason() returned a reason for an unrelated error") + } + if !IsRetryable(status.Error(codes.Unavailable, "down")) || IsRetryable(nil) { + t.Fatal("IsRetryable mismatch") + } + if !IsTerminal(status.Error(codes.InvalidArgument, "bad")) || IsTerminal(nil) { + t.Fatal("IsTerminal mismatch") + } + if !IsUnsupported(ErrUnsupportedProtocol) || IsUnsupported(errors.New("other")) { + t.Fatal("IsUnsupported mismatch") + } +} diff --git a/internal/kouja/fake_test.go b/internal/kouja/fake_test.go new file mode 100644 index 0000000..fdb398c --- /dev/null +++ b/internal/kouja/fake_test.go @@ -0,0 +1,58 @@ +package kouja + +import ( + "context" + + koujav1 "github.com/voidstarit/kouja/gen/kouja/v1" + "google.golang.org/grpc" +) + +type fakeAPI struct { + create func(context.Context, *koujav1.CreateVolumeRequest) (*koujav1.CreateVolumeResponse, error) + getVolume func(context.Context, *koujav1.GetVolumeRequest) (*koujav1.GetVolumeResponse, error) + getCredential func(context.Context, *koujav1.GetCredentialRequest) (*koujav1.GetCredentialResponse, error) + unexport func(context.Context, *koujav1.UnexportVolumeRequest) (*koujav1.UnexportVolumeResponse, error) + delete func(context.Context, *koujav1.DeleteVolumeRequest) (*koujav1.DeleteVolumeResponse, error) + getInfo func(context.Context, *koujav1.GetInfoRequest) (*koujav1.GetInfoResponse, error) + listClasses func(context.Context, *koujav1.ListStorageClassesRequest) (*koujav1.ListStorageClassesResponse, error) +} + +func (f *fakeAPI) CreateVolume(ctx context.Context, in *koujav1.CreateVolumeRequest, _ ...grpc.CallOption) (*koujav1.CreateVolumeResponse, error) { + return f.create(ctx, in) +} + +func (f *fakeAPI) GetVolume(ctx context.Context, in *koujav1.GetVolumeRequest, _ ...grpc.CallOption) (*koujav1.GetVolumeResponse, error) { + return f.getVolume(ctx, in) +} + +func (f *fakeAPI) GetCredential(ctx context.Context, in *koujav1.GetCredentialRequest, _ ...grpc.CallOption) (*koujav1.GetCredentialResponse, error) { + return f.getCredential(ctx, in) +} + +func (f *fakeAPI) UnexportVolume(ctx context.Context, in *koujav1.UnexportVolumeRequest, _ ...grpc.CallOption) (*koujav1.UnexportVolumeResponse, error) { + if f.unexport == nil { + return &koujav1.UnexportVolumeResponse{}, nil + } + return f.unexport(ctx, in) +} + +func (f *fakeAPI) DeleteVolume(ctx context.Context, in *koujav1.DeleteVolumeRequest, _ ...grpc.CallOption) (*koujav1.DeleteVolumeResponse, error) { + if f.delete == nil { + return &koujav1.DeleteVolumeResponse{}, nil + } + return f.delete(ctx, in) +} + +func (f *fakeAPI) GetInfo(ctx context.Context, in *koujav1.GetInfoRequest, _ ...grpc.CallOption) (*koujav1.GetInfoResponse, error) { + if f.getInfo == nil { + return &koujav1.GetInfoResponse{}, nil + } + return f.getInfo(ctx, in) +} + +func (f *fakeAPI) ListStorageClasses(ctx context.Context, in *koujav1.ListStorageClassesRequest, _ ...grpc.CallOption) (*koujav1.ListStorageClassesResponse, error) { + if f.listClasses == nil { + return &koujav1.ListStorageClassesResponse{}, nil + } + return f.listClasses(ctx, in) +} diff --git a/internal/kouja/health.go b/internal/kouja/health.go new file mode 100644 index 0000000..d1133be --- /dev/null +++ b/internal/kouja/health.go @@ -0,0 +1,156 @@ +package kouja + +import ( + "context" + "fmt" + "strings" + "time" + + koujav1 "github.com/voidstarit/kouja/gen/kouja/v1" + + "github.com/voidstarit/rookery/internal/blockdev" + "github.com/voidstarit/rookery/internal/store" +) + +// Health is the control-plane health of a kouja backend. +type Health struct { + Reachable bool + Ready bool + Detail string + Protocols []string + Classes []ClassCapacity + Owner string + CheckedAt time.Time +} + +// ClassCapacity describes one kouja storage class. +type ClassCapacity struct { + Name string + VGName string + State string + TotalBytes int64 + FreeBytes int64 + AllocatableBytes int64 + MaxVolumeBytes int64 +} + +// Checker performs on-demand kouja health checks. +type Checker struct { + API API + Now func() time.Time +} + +// Check probes the kouja endpoint, configured class, and optionally an +// existing disk to detect owner rotation. +func (c *Checker) Check(ctx context.Context, cfg Config, sample *store.Disk) (Health, error) { + if c == nil || c.API == nil { + return Health{}, fmt.Errorf("kouja client is not configured") + } + now := c.now() + health := Health{CheckedAt: now} + info, err := c.API.GetInfo(ctx, &koujav1.GetInfoRequest{}) + if err != nil { + health.Detail = err.Error() + return health, nil + } + health.Reachable = true + health.Owner = cfg.Owner + health.Protocols = protocolNames(info.GetProtocols()) + if !info.GetReady() { + health.Detail = "kouja agent is not ready" + return health, nil + } + if !containsProtocol(info.GetProtocols(), toKoujaProtocol(cfg.Protocol)) { + health.Detail = fmt.Sprintf("kouja agent does not advertise %s", cfg.Protocol) + return health, nil + } + classes, err := c.API.ListStorageClasses(ctx, &koujav1.ListStorageClassesRequest{}) + if err != nil { + health.Detail = err.Error() + return health, nil + } + var selected *koujav1.StorageClass + for _, class := range classes.GetClasses() { + if class.GetName() == cfg.StorageClass { + selected = class + break + } + } + if selected == nil { + health.Detail = fmt.Sprintf("kouja storage class %q does not exist", cfg.StorageClass) + return health, nil + } + health.Classes = append(health.Classes, ClassCapacity{ + Name: selected.GetName(), + VGName: selected.GetVgName(), + State: selected.GetVgState(), + TotalBytes: selected.GetTotalBytes(), + FreeBytes: selected.GetFreeBytes(), + AllocatableBytes: selected.GetAllocatableBytes(), + MaxVolumeBytes: selected.GetMaxVolumeBytes(), + }) + if sample != nil { + volumeID := strings.TrimSpace(sample.BackendMetadata.Volume.VolumeID) + if volumeID == "" { + health.Detail = "sampled disk has no kouja volume ID" + return health, nil + } + volume, err := c.API.GetVolume(ctx, &koujav1.GetVolumeRequest{VolumeId: volumeID}) + if err != nil { + health.Detail = err.Error() + return health, nil + } + if volume.GetDesired() == nil { + health.Detail = "sampled disk volume is missing" + return health, nil + } + if owner := strings.TrimSpace(sample.BackendMetadata.Volume.Owner); owner != "" && + volume.GetDesired() != nil && volume.GetDesired().GetOwner() != "" && + owner != volume.GetDesired().GetOwner() { + health.Detail = fmt.Sprintf("kouja volume owner changed from %q to %q", owner, volume.GetDesired().GetOwner()) + return health, nil + } + credential, err := c.API.GetCredential(ctx, &koujav1.GetCredentialRequest{VolumeId: volumeID}) + if err != nil { + health.Detail = err.Error() + return health, nil + } + if credential.GetProtocol() != toKoujaProtocol(cfg.Protocol) || credential.GetCredential() == "" { + health.Detail = "sampled disk credential is unavailable" + return health, nil + } + } + health.Ready = true + return health, nil +} + +func (c *Checker) now() time.Time { + if c != nil && c.Now != nil { + return c.Now().UTC() + } + return time.Now().UTC() +} + +func protocolNames(protocols []koujav1.Protocol) []string { + names := make([]string, 0, len(protocols)) + for _, value := range protocols { + switch value { + case koujav1.Protocol_PROTOCOL_ISCSI: + names = append(names, string(blockdev.ProtocolISCSI)) + case koujav1.Protocol_PROTOCOL_NVME_TCP: + names = append(names, string(blockdev.ProtocolNVMeTCP)) + default: + names = append(names, value.String()) + } + } + return names +} + +func containsProtocol(protocols []koujav1.Protocol, want koujav1.Protocol) bool { + for _, value := range protocols { + if value == want { + return true + } + } + return false +} diff --git a/internal/kouja/health_test.go b/internal/kouja/health_test.go new file mode 100644 index 0000000..8213d71 --- /dev/null +++ b/internal/kouja/health_test.go @@ -0,0 +1,171 @@ +package kouja + +import ( + "context" + "errors" + "testing" + "time" + + koujav1 "github.com/voidstarit/kouja/gen/kouja/v1" + + "github.com/voidstarit/rookery/internal/blockdev" + "github.com/voidstarit/rookery/internal/store" +) + +func healthInfo() *koujav1.GetInfoResponse { + return &koujav1.GetInfoResponse{ + Ready: true, + Protocols: []koujav1.Protocol{koujav1.Protocol_PROTOCOL_ISCSI}, + } +} + +func healthClasses() *koujav1.ListStorageClassesResponse { + return &koujav1.ListStorageClassesResponse{Classes: []*koujav1.StorageClass{{ + Name: "nvme", VgName: "vg0", VgState: "ready", + TotalBytes: 100, FreeBytes: 60, AllocatableBytes: 50, MaxVolumeBytes: 20, + }}} +} + +func healthSample() *store.Disk { + disk := testDisk(blockdev.ProtocolISCSI) + disk.BackendMetadata.Volume.Owner = "spiffe://kouja/caller/rookery" + return &disk +} + +func TestCheckerSuccess(t *testing.T) { + now := time.Now().UTC() + api := &fakeAPI{ + getInfo: func(context.Context, *koujav1.GetInfoRequest) (*koujav1.GetInfoResponse, error) { + return healthInfo(), nil + }, + listClasses: func(context.Context, *koujav1.ListStorageClassesRequest) (*koujav1.ListStorageClassesResponse, error) { + return healthClasses(), nil + }, + getVolume: func(context.Context, *koujav1.GetVolumeRequest) (*koujav1.GetVolumeResponse, error) { + return &koujav1.GetVolumeResponse{Desired: &koujav1.Volume{ + Owner: "spiffe://kouja/caller/rookery", + }}, nil + }, + getCredential: func(context.Context, *koujav1.GetCredentialRequest) (*koujav1.GetCredentialResponse, error) { + return &koujav1.GetCredentialResponse{Protocol: koujav1.Protocol_PROTOCOL_ISCSI, Credential: "secret"}, nil + }, + } + checker := &Checker{API: api, Now: func() time.Time { return now }} + health, err := checker.Check(context.Background(), provisionerConfig(blockdev.ProtocolISCSI), healthSample()) + if err != nil { + t.Fatalf("Check() = %v", err) + } + if !health.Reachable || !health.Ready || health.Owner == "" || + len(health.Classes) != 1 || health.Classes[0].AllocatableBytes != 50 { + t.Fatalf("health = %+v", health) + } + if len(health.Protocols) != 1 || health.Protocols[0] != "iscsi" { + t.Fatalf("protocols = %v", health.Protocols) + } +} + +func TestCheckerErrors(t *testing.T) { + apiBase := func() *fakeAPI { + return &fakeAPI{ + getInfo: func(context.Context, *koujav1.GetInfoRequest) (*koujav1.GetInfoResponse, error) { + return healthInfo(), nil + }, + listClasses: func(context.Context, *koujav1.ListStorageClassesRequest) (*koujav1.ListStorageClassesResponse, error) { + return healthClasses(), nil + }, + } + } + tests := []struct { + name string + api API + cfg Config + samp *store.Disk + }{ + {name: "nil api", api: nil, cfg: provisionerConfig(blockdev.ProtocolISCSI)}, + {name: "get info", api: &fakeAPI{getInfo: func(context.Context, *koujav1.GetInfoRequest) (*koujav1.GetInfoResponse, error) { + return nil, errors.New("boom") + }}, cfg: provisionerConfig(blockdev.ProtocolISCSI)}, + {name: "not ready", api: &fakeAPI{getInfo: func(context.Context, *koujav1.GetInfoRequest) (*koujav1.GetInfoResponse, error) { + return &koujav1.GetInfoResponse{Ready: false}, nil + }}, cfg: provisionerConfig(blockdev.ProtocolISCSI)}, + {name: "protocol missing", api: &fakeAPI{getInfo: func(context.Context, *koujav1.GetInfoRequest) (*koujav1.GetInfoResponse, error) { + return &koujav1.GetInfoResponse{Ready: true, Protocols: []koujav1.Protocol{koujav1.Protocol_PROTOCOL_NVME_TCP}}, nil + }}, cfg: provisionerConfig(blockdev.ProtocolISCSI)}, + {name: "list classes", api: &fakeAPI{ + getInfo: func(context.Context, *koujav1.GetInfoRequest) (*koujav1.GetInfoResponse, error) { + return healthInfo(), nil + }, + listClasses: func(context.Context, *koujav1.ListStorageClassesRequest) (*koujav1.ListStorageClassesResponse, error) { + return nil, errors.New("boom") + }, + }, cfg: provisionerConfig(blockdev.ProtocolISCSI)}, + {name: "class missing", api: apiBase(), cfg: Config{Protocol: blockdev.ProtocolISCSI, StorageClass: "missing"}}, + {name: "sample missing volume", api: apiBase(), cfg: provisionerConfig(blockdev.ProtocolISCSI), samp: &store.Disk{}}, + {name: "sample get volume", api: func() *fakeAPI { + value := apiBase() + value.getVolume = func(context.Context, *koujav1.GetVolumeRequest) (*koujav1.GetVolumeResponse, error) { + return nil, errors.New("boom") + } + return value + }(), cfg: provisionerConfig(blockdev.ProtocolISCSI), samp: healthSample()}, + {name: "sample owner", api: func() *fakeAPI { + value := apiBase() + value.getVolume = func(context.Context, *koujav1.GetVolumeRequest) (*koujav1.GetVolumeResponse, error) { + return &koujav1.GetVolumeResponse{Desired: &koujav1.Volume{Owner: "spiffe://kouja/caller/other"}}, nil + } + return value + }(), cfg: provisionerConfig(blockdev.ProtocolISCSI), samp: healthSample()}, + {name: "sample credential", api: func() *fakeAPI { + value := apiBase() + value.getVolume = func(context.Context, *koujav1.GetVolumeRequest) (*koujav1.GetVolumeResponse, error) { + return &koujav1.GetVolumeResponse{Desired: &koujav1.Volume{Owner: "spiffe://kouja/caller/rookery"}}, nil + } + value.getCredential = func(context.Context, *koujav1.GetCredentialRequest) (*koujav1.GetCredentialResponse, error) { + return nil, errors.New("boom") + } + return value + }(), cfg: provisionerConfig(blockdev.ProtocolISCSI), samp: healthSample()}, + {name: "sample empty credential", api: func() *fakeAPI { + value := apiBase() + value.getVolume = func(context.Context, *koujav1.GetVolumeRequest) (*koujav1.GetVolumeResponse, error) { + return &koujav1.GetVolumeResponse{Desired: &koujav1.Volume{Owner: "spiffe://kouja/caller/rookery"}}, nil + } + value.getCredential = func(context.Context, *koujav1.GetCredentialRequest) (*koujav1.GetCredentialResponse, error) { + return &koujav1.GetCredentialResponse{Protocol: koujav1.Protocol_PROTOCOL_ISCSI}, nil + } + return value + }(), cfg: provisionerConfig(blockdev.ProtocolISCSI), samp: healthSample()}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + checker := &Checker{API: tt.api, Now: func() time.Time { return time.Now().UTC() }} + health, err := checker.Check(context.Background(), tt.cfg, tt.samp) + if tt.name == "nil api" { + if err == nil { + t.Fatal("Check() accepted nil API") + } + return + } + if err != nil { + t.Fatal(err) + } + if health.Ready { + t.Fatalf("health unexpectedly ready: %+v", health) + } + }) + } +} + +func TestProtocolNames(t *testing.T) { + names := protocolNames([]koujav1.Protocol{ + koujav1.Protocol_PROTOCOL_ISCSI, + koujav1.Protocol_PROTOCOL_NVME_TCP, + koujav1.Protocol_PROTOCOL_UNSPECIFIED, + }) + if len(names) != 3 || names[0] != "iscsi" || names[1] != "nvme-tcp" { + t.Fatalf("protocolNames() = %v", names) + } + if !containsProtocol([]koujav1.Protocol{koujav1.Protocol_PROTOCOL_ISCSI}, koujav1.Protocol_PROTOCOL_ISCSI) { + t.Fatal("containsProtocol() = false") + } +} diff --git a/internal/kouja/preparer.go b/internal/kouja/preparer.go new file mode 100644 index 0000000..559925f --- /dev/null +++ b/internal/kouja/preparer.go @@ -0,0 +1,184 @@ +package kouja + +import ( + "context" + "errors" + "fmt" + "strings" + "time" + + koujav1 "github.com/voidstarit/kouja/gen/kouja/v1" + + "github.com/voidstarit/rookery/internal/blockdev" + "github.com/voidstarit/rookery/internal/store" +) + +var ( + // ErrExportChanged means the live export no longer matches the stored reference. + ErrExportChanged = errors.New("kouja export changed") + // ErrStaleObservation means a fresh observation shows a missing target or ACL. + ErrStaleObservation = errors.New("kouja export observation is stale or incomplete") + // ErrSessionActive means another initiator currently owns the export. + ErrSessionActive = errors.New("kouja export has an active session") + // ErrMissingCredential means kouja returned no credential. + ErrMissingCredential = errors.New("kouja returned an empty credential") +) + +// ObservationFreshness is the maximum age for an observation to be trusted. +const ObservationFreshness = 2 * time.Minute + +// Preparer fetches the live export and credential for a VM start. +type Preparer struct { + API API + Now func() time.Time +} + +// Prepare returns a runtime attachment for disk. +func (p *Preparer) Prepare(ctx context.Context, cfg Config, disk store.Disk) (blockdev.Attachment, error) { + if p == nil || p.API == nil { + return blockdev.Attachment{}, errors.New("kouja client is not configured") + } + protocol := blockdev.Protocol(disk.Protocol) + if protocol == "" { + protocol = cfg.Protocol + } + if err := validateProtocol(protocol); err != nil { + return blockdev.Attachment{}, err + } + volumeID := strings.TrimSpace(disk.BackendMetadata.Volume.VolumeID) + if volumeID == "" { + return blockdev.Attachment{}, errors.New("disk has no kouja volume ID") + } + response, err := p.API.GetVolume(ctx, &koujav1.GetVolumeRequest{VolumeId: volumeID}) + if err != nil { + return blockdev.Attachment{}, err + } + if err := p.validateLiveExport(cfg, protocol, disk, response); err != nil { + return blockdev.Attachment{}, err + } + credential, err := p.API.GetCredential(ctx, &koujav1.GetCredentialRequest{VolumeId: volumeID}) + if err != nil { + return blockdev.Attachment{}, err + } + if credential.GetProtocol() != toKoujaProtocol(protocol) { + return blockdev.Attachment{}, fmt.Errorf("kouja credential protocol is %s", credential.GetProtocol()) + } + if credential.GetCredential() == "" { + return blockdev.Attachment{}, ErrMissingCredential + } + resourceID := strings.TrimSpace(disk.BackendMetadata.Volume.ResourceID) + if resourceID == "" { + return blockdev.Attachment{}, errors.New("disk has no kouja resource ID") + } + reference := disk.BackendMetadata.Attachment + attachment := blockdev.Attachment{Protocol: protocol} + switch protocol { + case blockdev.ProtocolISCSI: + if reference.ISCSI == nil { + return blockdev.Attachment{}, errors.New("disk has no iSCSI reference") + } + value := *reference.ISCSI + attachment.ISCSI = &blockdev.ISCSIAttachment{ + ISCSIReference: value, + Username: "kouja-" + resourceID, + Credential: credential.GetCredential(), + } + case blockdev.ProtocolNVMeTCP: + if reference.NVMeTCP == nil { + return blockdev.Attachment{}, errors.New("disk has no NVMe/TCP reference") + } + value := *reference.NVMeTCP + attachment.NVMeTCP = &blockdev.NVMeTCPAttachment{ + NVMeTCPReference: value, + PSK: credential.GetCredential(), + } + } + if err := attachment.ValidateComplete(); err != nil { + return blockdev.Attachment{}, err + } + return attachment, nil +} + +func (p *Preparer) validateLiveExport(cfg Config, protocol blockdev.Protocol, disk store.Disk, response *koujav1.GetVolumeResponse) error { + desired := response.GetDesired() + export := response.GetExport() + if desired == nil || export == nil { + return errors.New("kouja get volume returned no desired volume or export") + } + if desired.GetState() != koujav1.VolumeState_VOLUME_STATE_AVAILABLE && + desired.GetState() != koujav1.VolumeState_VOLUME_STATE_EXPORTED { + return fmt.Errorf("kouja volume state is %s", desired.GetState()) + } + if export.GetState() != koujav1.ExportState_EXPORT_STATE_EXPORTED { + return fmt.Errorf("kouja export state is %s", export.GetState()) + } + if export.GetProtocol() != toKoujaProtocol(protocol) { + return fmt.Errorf("kouja export protocol is %s", export.GetProtocol()) + } + observed := response.GetObserved() + observeAt := time.Time{} + if observed != nil && observed.GetObservedAt() != nil { + observeAt = observed.GetObservedAt().AsTime() + } + now := p.now() + if !observeAt.IsZero() && now.Sub(observeAt) <= ObservationFreshness { + if !observed.GetTargetPresent() || !observed.GetAclPresent() { + return ErrStaleObservation + } + } + if export.GetSessionState() == koujav1.SessionState_SESSION_STATE_ACTIVE { + return ErrSessionActive + } + activeSessions := int32(0) + if observed != nil { + activeSessions = observed.GetActiveSessions() + } + if (export.GetSessionState() == koujav1.SessionState_SESSION_STATE_UNKNOWN || + export.GetSessionState() == koujav1.SessionState_SESSION_STATE_UNSPECIFIED) && activeSessions > 0 { + return ErrSessionActive + } + if owner := strings.TrimSpace(disk.BackendMetadata.Volume.Owner); owner != "" && + desired.GetOwner() != "" && owner != desired.GetOwner() { + return fmt.Errorf("kouja volume owner changed from %q to %q", owner, desired.GetOwner()) + } + reference := disk.BackendMetadata.Attachment + switch protocol { + case blockdev.ProtocolISCSI: + if reference.ISCSI == nil { + return ErrExportChanged + } + exportPortal := reference.ISCSI.ExportPortal + if exportPortal == "" { + exportPortal = reference.ISCSI.Portal + } + if export.GetPortal() != exportPortal || + export.GetTargetName() != reference.ISCSI.Target || + int(export.GetLun()) != reference.ISCSI.LUN || + (cfg.Initiator != "" && export.GetInitiator() != "" && export.GetInitiator() != cfg.Initiator) { + return ErrExportChanged + } + case blockdev.ProtocolNVMeTCP: + if reference.NVMeTCP == nil { + return ErrExportChanged + } + exportAddress := reference.NVMeTCP.ExportAddress + if exportAddress == "" { + exportAddress = reference.NVMeTCP.Address + } + if export.GetPortal() != exportAddress || + export.GetTargetName() != reference.NVMeTCP.NQN || + int(export.GetNsid()) != reference.NVMeTCP.NSID || + (reference.NVMeTCP.HostNQN != "" && export.GetInitiator() != "" && + export.GetInitiator() != reference.NVMeTCP.HostNQN) { + return ErrExportChanged + } + } + return nil +} + +func (p *Preparer) now() time.Time { + if p != nil && p.Now != nil { + return p.Now().UTC() + } + return time.Now().UTC() +} diff --git a/internal/kouja/preparer_test.go b/internal/kouja/preparer_test.go new file mode 100644 index 0000000..e52e48e --- /dev/null +++ b/internal/kouja/preparer_test.go @@ -0,0 +1,214 @@ +package kouja + +import ( + "context" + "errors" + "testing" + "time" + + koujav1 "github.com/voidstarit/kouja/gen/kouja/v1" + "google.golang.org/protobuf/types/known/timestamppb" + + "github.com/voidstarit/rookery/internal/blockdev" + "github.com/voidstarit/rookery/internal/store" +) + +func testDisk(protocol blockdev.Protocol) store.Disk { + reference := blockdev.Reference{Protocol: protocol} + if protocol == blockdev.ProtocolISCSI { + reference.ISCSI = &blockdev.ISCSIReference{ + Portal: "10.0.0.1:3260", ExportPortal: "10.0.0.1:3260", + Target: "iqn.2026-11.io.example:disk", LUN: 0, + Initiator: "iqn.2026-11.io.example:rookery", + } + } else { + reference.NVMeTCP = &blockdev.NVMeTCPReference{ + Address: "10.0.0.1:4420", ExportAddress: "10.0.0.1:4420", + NQN: "nqn.2026-11.io.example:disk", HostNQN: "nqn.2026-11.io.example:rookery", NSID: 1, + } + } + return store.Disk{ + Protocol: string(protocol), + BackendMetadata: store.DiskBackendMetadata{ + Volume: store.VolumeMetadata{ + VolumeID: "volume-1", ResourceID: "resource-1", Owner: "spiffe://kouja/caller/rookery", + }, + Attachment: reference, + }, + } +} + +func testGetVolumeResponse(protocol blockdev.Protocol, observed *koujav1.Observation) *koujav1.GetVolumeResponse { + export := &koujav1.Export{ + Protocol: toKoujaProtocol(protocol), TargetName: "iqn.2026-11.io.example:disk", + Portal: "10.0.0.1:3260", Lun: 0, Nsid: 1, + State: koujav1.ExportState_EXPORT_STATE_EXPORTED, + } + if protocol == blockdev.ProtocolNVMeTCP { + export.TargetName = "nqn.2026-11.io.example:disk" + export.Portal = "10.0.0.1:4420" + } + return &koujav1.GetVolumeResponse{ + Desired: &koujav1.Volume{ + Owner: "spiffe://kouja/caller/rookery", VolumeId: "volume-1", + ResourceUuid: "resource-1", State: koujav1.VolumeState_VOLUME_STATE_EXPORTED, + }, + Observed: observed, + Export: export, + } +} + +func testObserved(now time.Time) *koujav1.Observation { + return &koujav1.Observation{ + TargetPresent: true, AclPresent: true, ActiveSessions: 0, + SessionState: koujav1.SessionState_SESSION_STATE_NONE, + ObservedAt: timestamppb.New(now), + } +} + +func TestPreparerSuccess(t *testing.T) { + now := time.Now().UTC() + for _, protocol := range []blockdev.Protocol{blockdev.ProtocolISCSI, blockdev.ProtocolNVMeTCP} { + t.Run(string(protocol), func(t *testing.T) { + api := &fakeAPI{ + getVolume: func(context.Context, *koujav1.GetVolumeRequest) (*koujav1.GetVolumeResponse, error) { + return testGetVolumeResponse(protocol, testObserved(now)), nil + }, + getCredential: func(context.Context, *koujav1.GetCredentialRequest) (*koujav1.GetCredentialResponse, error) { + return &koujav1.GetCredentialResponse{Protocol: toKoujaProtocol(protocol), Credential: "secret"}, nil + }, + } + preparer := &Preparer{API: api, Now: func() time.Time { return now }} + attachment, err := preparer.Prepare(context.Background(), provisionerConfig(protocol), testDisk(protocol)) + if err != nil { + t.Fatalf("Prepare() = %v", err) + } + if protocol == blockdev.ProtocolISCSI { + if attachment.ISCSI == nil || attachment.ISCSI.Username != "kouja-resource-1" || + attachment.ISCSI.Credential != "secret" { + t.Fatalf("iSCSI attachment = %+v", attachment) + } + } else if attachment.NVMeTCP == nil || attachment.NVMeTCP.PSK != "secret" { + t.Fatalf("NVMe attachment = %+v", attachment) + } + }) + } +} + +func TestPreparerErrors(t *testing.T) { + now := time.Now().UTC() + disk := testDisk(blockdev.ProtocolISCSI) + cfg := provisionerConfig(blockdev.ProtocolISCSI) + good := testGetVolumeResponse(blockdev.ProtocolISCSI, testObserved(now)) + tests := []struct { + name string + api API + disk store.Disk + cfg Config + }{ + {name: "nil api", disk: disk, cfg: cfg}, + {name: "protocol", api: &fakeAPI{}, disk: store.Disk{Protocol: string(blockdev.ProtocolLocal)}, cfg: cfg}, + {name: "missing volume", api: &fakeAPI{}, disk: store.Disk{Protocol: string(blockdev.ProtocolISCSI)}, cfg: cfg}, + {name: "get volume", api: &fakeAPI{getVolume: func(context.Context, *koujav1.GetVolumeRequest) (*koujav1.GetVolumeResponse, error) { + return nil, errors.New("boom") + }}, disk: disk, cfg: cfg}, + {name: "nil response", api: &fakeAPI{getVolume: func(context.Context, *koujav1.GetVolumeRequest) (*koujav1.GetVolumeResponse, error) { + return &koujav1.GetVolumeResponse{}, nil + }}, disk: disk, cfg: cfg}, + {name: "volume state", api: &fakeAPI{getVolume: func(context.Context, *koujav1.GetVolumeRequest) (*koujav1.GetVolumeResponse, error) { + value := testGetVolumeResponse(blockdev.ProtocolISCSI, testObserved(now)) + value.Desired.State = koujav1.VolumeState_VOLUME_STATE_FAILED + return value, nil + }}, disk: disk, cfg: cfg}, + {name: "export state", api: &fakeAPI{getVolume: func(context.Context, *koujav1.GetVolumeRequest) (*koujav1.GetVolumeResponse, error) { + value := testGetVolumeResponse(blockdev.ProtocolISCSI, testObserved(now)) + value.Export.State = koujav1.ExportState_EXPORT_STATE_FAILED + return value, nil + }}, disk: disk, cfg: cfg}, + {name: "export protocol", api: &fakeAPI{getVolume: func(context.Context, *koujav1.GetVolumeRequest) (*koujav1.GetVolumeResponse, error) { + value := testGetVolumeResponse(blockdev.ProtocolISCSI, testObserved(now)) + value.Export.Protocol = koujav1.Protocol_PROTOCOL_NVME_TCP + return value, nil + }}, disk: disk, cfg: cfg}, + {name: "fresh missing target", api: &fakeAPI{getVolume: func(context.Context, *koujav1.GetVolumeRequest) (*koujav1.GetVolumeResponse, error) { + observed := testObserved(now) + observed.TargetPresent = false + return testGetVolumeResponse(blockdev.ProtocolISCSI, observed), nil + }}, disk: disk, cfg: cfg}, + {name: "active session", api: &fakeAPI{getVolume: func(context.Context, *koujav1.GetVolumeRequest) (*koujav1.GetVolumeResponse, error) { + value := testGetVolumeResponse(blockdev.ProtocolISCSI, testObserved(now)) + value.Export.SessionState = koujav1.SessionState_SESSION_STATE_ACTIVE + return value, nil + }}, disk: disk, cfg: cfg}, + {name: "unknown active session", api: &fakeAPI{getVolume: func(context.Context, *koujav1.GetVolumeRequest) (*koujav1.GetVolumeResponse, error) { + value := testGetVolumeResponse(blockdev.ProtocolISCSI, testObserved(now)) + value.Export.SessionState = koujav1.SessionState_SESSION_STATE_UNKNOWN + value.Observed.ActiveSessions = 1 + return value, nil + }}, disk: disk, cfg: cfg}, + {name: "owner", api: &fakeAPI{getVolume: func(context.Context, *koujav1.GetVolumeRequest) (*koujav1.GetVolumeResponse, error) { + value := testGetVolumeResponse(blockdev.ProtocolISCSI, testObserved(now)) + value.Desired.Owner = "spiffe://kouja/caller/other" + return value, nil + }}, disk: disk, cfg: cfg}, + {name: "missing reference", api: &fakeAPI{getVolume: func(context.Context, *koujav1.GetVolumeRequest) (*koujav1.GetVolumeResponse, error) { + return good, nil + }}, disk: store.Disk{Protocol: string(blockdev.ProtocolISCSI), BackendMetadata: store.DiskBackendMetadata{ + Volume: store.VolumeMetadata{VolumeID: "volume-1", ResourceID: "resource-1"}, + Attachment: blockdev.Reference{Protocol: blockdev.ProtocolISCSI}, + }}, cfg: cfg}, + {name: "export portal changed", api: &fakeAPI{getVolume: func(context.Context, *koujav1.GetVolumeRequest) (*koujav1.GetVolumeResponse, error) { + value := testGetVolumeResponse(blockdev.ProtocolISCSI, testObserved(now)) + value.Export.Portal = "10.0.0.2:3260" + return value, nil + }}, disk: disk, cfg: cfg}, + {name: "target changed", api: &fakeAPI{getVolume: func(context.Context, *koujav1.GetVolumeRequest) (*koujav1.GetVolumeResponse, error) { + value := testGetVolumeResponse(blockdev.ProtocolISCSI, testObserved(now)) + value.Export.TargetName = "iqn.other" + return value, nil + }}, disk: disk, cfg: cfg}, + {name: "credential error", api: &fakeAPI{ + getVolume: func(context.Context, *koujav1.GetVolumeRequest) (*koujav1.GetVolumeResponse, error) { + return good, nil + }, + getCredential: func(context.Context, *koujav1.GetCredentialRequest) (*koujav1.GetCredentialResponse, error) { + return nil, errors.New("boom") + }, + }, disk: disk, cfg: cfg}, + {name: "credential protocol", api: &fakeAPI{ + getVolume: func(context.Context, *koujav1.GetVolumeRequest) (*koujav1.GetVolumeResponse, error) { + return good, nil + }, + getCredential: func(context.Context, *koujav1.GetCredentialRequest) (*koujav1.GetCredentialResponse, error) { + return &koujav1.GetCredentialResponse{Protocol: koujav1.Protocol_PROTOCOL_NVME_TCP, Credential: "secret"}, nil + }, + }, disk: disk, cfg: cfg}, + {name: "empty credential", api: &fakeAPI{ + getVolume: func(context.Context, *koujav1.GetVolumeRequest) (*koujav1.GetVolumeResponse, error) { + return good, nil + }, + getCredential: func(context.Context, *koujav1.GetCredentialRequest) (*koujav1.GetCredentialResponse, error) { + return &koujav1.GetCredentialResponse{Protocol: koujav1.Protocol_PROTOCOL_ISCSI}, nil + }, + }, disk: disk, cfg: cfg}, + {name: "missing resource", api: &fakeAPI{ + getVolume: func(context.Context, *koujav1.GetVolumeRequest) (*koujav1.GetVolumeResponse, error) { + return good, nil + }, + getCredential: func(context.Context, *koujav1.GetCredentialRequest) (*koujav1.GetCredentialResponse, error) { + return &koujav1.GetCredentialResponse{Protocol: koujav1.Protocol_PROTOCOL_ISCSI, Credential: "secret"}, nil + }, + }, disk: store.Disk{Protocol: string(blockdev.ProtocolISCSI), BackendMetadata: store.DiskBackendMetadata{ + Volume: store.VolumeMetadata{VolumeID: "volume-1"}, + Attachment: disk.BackendMetadata.Attachment, + }}, cfg: cfg}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + preparer := &Preparer{API: tt.api, Now: func() time.Time { return now }} + if _, err := preparer.Prepare(context.Background(), tt.cfg, tt.disk); err == nil { + t.Fatal("Prepare() accepted invalid input") + } + }) + } +} diff --git a/internal/kouja/provisioner.go b/internal/kouja/provisioner.go new file mode 100644 index 0000000..1d2e788 --- /dev/null +++ b/internal/kouja/provisioner.go @@ -0,0 +1,181 @@ +package kouja + +import ( + "context" + "errors" + "fmt" + "strings" + + "github.com/google/uuid" + koujacli "github.com/voidstarit/kouja/gen/client" + koujav1 "github.com/voidstarit/kouja/gen/kouja/v1" + "google.golang.org/protobuf/proto" + + "github.com/voidstarit/rookery/internal/blockdev" +) + +// CreateRequest identifies a volume creation request. +type CreateRequest struct { + VolumeID string + RequestID string + SizeBytes int64 +} + +// CreatedVolume is the secret-free result of a kouja volume create. +type CreatedVolume struct { + Reference blockdev.Reference + Owner string + ResourceID string + ActualSizeBytes uint64 +} + +// Provisioner creates and exports kouja volumes. +type Provisioner struct { + API API +} + +// Create creates an exported volume and returns its persisted reference. +func (p *Provisioner) Create(ctx context.Context, cfg Config, request CreateRequest) (CreatedVolume, error) { + if p == nil || p.API == nil { + return CreatedVolume{}, errors.New("kouja client is not configured") + } + if err := validateProtocol(cfg.Protocol); err != nil { + return CreatedVolume{}, err + } + request.VolumeID = strings.TrimSpace(request.VolumeID) + if request.VolumeID == "" { + return CreatedVolume{}, errors.New("kouja volume ID is required") + } + if request.SizeBytes < 1 { + return CreatedVolume{}, errors.New("kouja volume size must be positive") + } + if strings.TrimSpace(request.RequestID) == "" { + request.RequestID = uuid.NewString() + } + ctx = koujacli.WithRequestID(ctx, request.RequestID) + initiator := cfg.Initiator + if cfg.Protocol == blockdev.ProtocolNVMeTCP { + initiator = cfg.HostNQN + } + response, err := p.API.CreateVolume(ctx, &koujav1.CreateVolumeRequest{ + VolumeId: request.VolumeID, + SizeBytes: request.SizeBytes, + Export: proto.Bool(true), + Protocol: toKoujaProtocol(cfg.Protocol), + Initiator: initiator, + Wipe: false, + RequestId: request.RequestID, + IncludeCredential: proto.Bool(false), + StorageClass: cfg.StorageClass, + }) + if err != nil { + return CreatedVolume{}, err + } + volume := response.GetVolume() + export := response.GetExport() + if volume == nil || export == nil { + return CreatedVolume{}, errors.New("kouja create returned no volume or export") + } + if volume.GetVolumeId() != request.VolumeID { + return CreatedVolume{}, fmt.Errorf("kouja created volume %q, want %q", volume.GetVolumeId(), request.VolumeID) + } + if strings.TrimSpace(volume.GetResourceUuid()) == "" { + return CreatedVolume{}, errors.New("kouja volume has no resource UUID") + } + if volume.GetState() != koujav1.VolumeState_VOLUME_STATE_AVAILABLE && + volume.GetState() != koujav1.VolumeState_VOLUME_STATE_EXPORTED { + return CreatedVolume{}, fmt.Errorf("kouja volume state is %s", volume.GetState()) + } + if export.GetState() != koujav1.ExportState_EXPORT_STATE_EXPORTED { + return CreatedVolume{}, fmt.Errorf("kouja export state is %s", export.GetState()) + } + if export.GetProtocol() != toKoujaProtocol(cfg.Protocol) { + return CreatedVolume{}, fmt.Errorf("kouja export protocol is %s", export.GetProtocol()) + } + if strings.TrimSpace(export.GetTargetName()) == "" || strings.TrimSpace(export.GetPortal()) == "" { + return CreatedVolume{}, errors.New("kouja export has no target or portal") + } + if export.GetLun() < 0 { + return CreatedVolume{}, errors.New("kouja export has a negative LUN") + } + if cfg.Protocol == blockdev.ProtocolNVMeTCP && export.GetNsid() < 0 { + return CreatedVolume{}, errors.New("kouja export has a negative NSID") + } + if strings.TrimSpace(volume.GetOwner()) == "" { + return CreatedVolume{}, errors.New("kouja volume has no owner") + } + effectivePortal := cfg.Portal + if effectivePortal == "" { + effectivePortal = export.GetPortal() + if err := validatePortal(effectivePortal); err != nil { + return CreatedVolume{}, err + } + } + reference, err := referenceFromExport(cfg, volume, export, effectivePortal) + if err != nil { + return CreatedVolume{}, err + } + if err := reference.ValidateComplete(); err != nil { + return CreatedVolume{}, err + } + return CreatedVolume{ + Reference: reference, + Owner: volume.GetOwner(), + ResourceID: volume.GetResourceUuid(), + ActualSizeBytes: uint64(maxInt64(volume.GetActualSizeBytes(), 0)), + }, nil +} + +func referenceFromExport(cfg Config, volume *koujav1.Volume, export *koujav1.Export, effectivePortal string) (blockdev.Reference, error) { + switch cfg.Protocol { + case blockdev.ProtocolISCSI: + return blockdev.Reference{ + Protocol: blockdev.ProtocolISCSI, + ISCSI: &blockdev.ISCSIReference{ + Portal: effectivePortal, + ExportPortal: export.GetPortal(), + Target: export.GetTargetName(), + LUN: int(export.GetLun()), + Initiator: cfg.Initiator, + }, + }, nil + case blockdev.ProtocolNVMeTCP: + return blockdev.Reference{ + Protocol: blockdev.ProtocolNVMeTCP, + NVMeTCP: &blockdev.NVMeTCPReference{ + Address: effectivePortal, + ExportAddress: export.GetPortal(), + NQN: export.GetTargetName(), + HostNQN: cfg.HostNQN, + NSID: int(export.GetNsid()), + }, + }, nil + default: + return blockdev.Reference{}, fmt.Errorf("%w: %s", ErrUnsupportedProtocol, cfg.Protocol) + } +} + +func validateProtocol(protocol blockdev.Protocol) error { + switch protocol { + case blockdev.ProtocolISCSI, blockdev.ProtocolNVMeTCP: + return nil + default: + return fmt.Errorf("%w: %s", ErrUnsupportedProtocol, protocol) + } +} + +func toKoujaProtocol(protocol blockdev.Protocol) koujav1.Protocol { + switch protocol { + case blockdev.ProtocolNVMeTCP: + return koujav1.Protocol_PROTOCOL_NVME_TCP + default: + return koujav1.Protocol_PROTOCOL_ISCSI + } +} + +func maxInt64(value, floor int64) int64 { + if value < floor { + return floor + } + return value +} diff --git a/internal/kouja/provisioner_test.go b/internal/kouja/provisioner_test.go new file mode 100644 index 0000000..7eecdeb --- /dev/null +++ b/internal/kouja/provisioner_test.go @@ -0,0 +1,170 @@ +package kouja + +import ( + "context" + "errors" + "testing" + + koujacli "github.com/voidstarit/kouja/gen/client" + koujav1 "github.com/voidstarit/kouja/gen/kouja/v1" + + "github.com/voidstarit/rookery/internal/blockdev" +) + +func provisionerConfig(protocol blockdev.Protocol) Config { + return Config{ + Protocol: protocol, + StorageClass: "nvme", + Initiator: "iqn.2026-11.io.example:rookery", + HostNQN: "nqn.2026-11.io.example:rookery", + Owner: "spiffe://kouja/caller/rookery", + } +} + +func createdResponse(protocol blockdev.Protocol) *koujav1.CreateVolumeResponse { + export := &koujav1.Export{ + Protocol: toKoujaProtocol(protocol), + TargetName: "iqn.2026-11.io.example:disk", + Portal: "10.0.0.1:3260", + Lun: 0, + Nsid: 1, + State: koujav1.ExportState_EXPORT_STATE_EXPORTED, + } + if protocol == blockdev.ProtocolNVMeTCP { + export.TargetName = "nqn.2026-11.io.example:disk" + export.Portal = "10.0.0.1:4420" + } + return &koujav1.CreateVolumeResponse{ + Volume: &koujav1.Volume{ + Owner: "spiffe://kouja/caller/rookery", + VolumeId: "volume-1", + ResourceUuid: "resource-1", + ActualSizeBytes: 10 << 30, + State: koujav1.VolumeState_VOLUME_STATE_EXPORTED, + }, + Export: export, + } +} + +func TestProvisionerCreate(t *testing.T) { + for _, protocol := range []blockdev.Protocol{blockdev.ProtocolISCSI, blockdev.ProtocolNVMeTCP} { + t.Run(string(protocol), func(t *testing.T) { + var gotRequest *koujav1.CreateVolumeRequest + api := &fakeAPI{create: func(ctx context.Context, in *koujav1.CreateVolumeRequest) (*koujav1.CreateVolumeResponse, error) { + gotRequest = in + if got := koujacli.RequestIDFromContext(ctx); got != "request-1" { + t.Fatalf("request ID in context = %q", got) + } + return createdResponse(protocol), nil + }} + provisioner := &Provisioner{API: api} + created, err := provisioner.Create(context.Background(), provisionerConfig(protocol), CreateRequest{ + VolumeID: "volume-1", RequestID: "request-1", SizeBytes: 10 << 30, + }) + if err != nil { + t.Fatalf("Create() = %v", err) + } + if gotRequest.GetProtocol() != toKoujaProtocol(protocol) || !gotRequest.GetExport() || + gotRequest.GetIncludeCredential() || gotRequest.GetStorageClass() != "nvme" { + t.Fatalf("request = %+v", gotRequest) + } + if created.Reference.Protocol != protocol || created.ResourceID != "resource-1" || + created.ActualSizeBytes != 10<<30 { + t.Fatalf("created = %+v", created) + } + if protocol == blockdev.ProtocolISCSI && created.Reference.ISCSI == nil { + t.Fatal("missing iSCSI reference") + } + if protocol == blockdev.ProtocolNVMeTCP && created.Reference.NVMeTCP == nil { + t.Fatal("missing NVMe reference") + } + }) + } +} + +func TestProvisionerErrors(t *testing.T) { + cfg := provisionerConfig(blockdev.ProtocolISCSI) + base := createdResponse(blockdev.ProtocolISCSI) + tests := []struct { + name string + api API + cfg Config + req CreateRequest + }{ + {name: "nil api", api: nil, cfg: cfg, req: CreateRequest{VolumeID: "v", SizeBytes: 1}}, + {name: "protocol", api: &fakeAPI{}, cfg: Config{Protocol: blockdev.ProtocolLocal}, req: CreateRequest{VolumeID: "v", SizeBytes: 1}}, + {name: "create error", api: &fakeAPI{create: func(context.Context, *koujav1.CreateVolumeRequest) (*koujav1.CreateVolumeResponse, error) { + return nil, errors.New("boom") + }}, cfg: cfg, req: CreateRequest{VolumeID: "v", SizeBytes: 1}}, + {name: "nil response", api: &fakeAPI{create: func(context.Context, *koujav1.CreateVolumeRequest) (*koujav1.CreateVolumeResponse, error) { + return &koujav1.CreateVolumeResponse{}, nil + }}, cfg: cfg, req: CreateRequest{VolumeID: "v", SizeBytes: 1}}, + {name: "volume id mismatch", api: &fakeAPI{create: func(context.Context, *koujav1.CreateVolumeRequest) (*koujav1.CreateVolumeResponse, error) { + return mutateCreated(base, func(r *koujav1.CreateVolumeResponse) { r.Volume.VolumeId = "other" }), nil + }}, cfg: cfg, req: CreateRequest{VolumeID: "v", SizeBytes: 1}}, + {name: "resource", api: &fakeAPI{create: func(context.Context, *koujav1.CreateVolumeRequest) (*koujav1.CreateVolumeResponse, error) { + return mutateCreated(base, func(r *koujav1.CreateVolumeResponse) { r.Volume.ResourceUuid = "" }), nil + }}, cfg: cfg, req: CreateRequest{VolumeID: "v", SizeBytes: 1}}, + {name: "state", api: &fakeAPI{create: func(context.Context, *koujav1.CreateVolumeRequest) (*koujav1.CreateVolumeResponse, error) { + return mutateCreated(base, func(r *koujav1.CreateVolumeResponse) { + r.Volume.State = koujav1.VolumeState_VOLUME_STATE_FAILED + }), nil + }}, cfg: cfg, req: CreateRequest{VolumeID: "v", SizeBytes: 1}}, + {name: "export state", api: &fakeAPI{create: func(context.Context, *koujav1.CreateVolumeRequest) (*koujav1.CreateVolumeResponse, error) { + return mutateCreated(base, func(r *koujav1.CreateVolumeResponse) { + r.Export.State = koujav1.ExportState_EXPORT_STATE_FAILED + }), nil + }}, cfg: cfg, req: CreateRequest{VolumeID: "v", SizeBytes: 1}}, + {name: "protocol", api: &fakeAPI{create: func(context.Context, *koujav1.CreateVolumeRequest) (*koujav1.CreateVolumeResponse, error) { + return mutateCreated(base, func(r *koujav1.CreateVolumeResponse) { + r.Export.Protocol = koujav1.Protocol_PROTOCOL_NVME_TCP + }), nil + }}, cfg: cfg, req: CreateRequest{VolumeID: "v", SizeBytes: 1}}, + {name: "target", api: &fakeAPI{create: func(context.Context, *koujav1.CreateVolumeRequest) (*koujav1.CreateVolumeResponse, error) { + return mutateCreated(base, func(r *koujav1.CreateVolumeResponse) { r.Export.TargetName = "" }), nil + }}, cfg: cfg, req: CreateRequest{VolumeID: "v", SizeBytes: 1}}, + {name: "portal", api: &fakeAPI{create: func(context.Context, *koujav1.CreateVolumeRequest) (*koujav1.CreateVolumeResponse, error) { + return mutateCreated(base, func(r *koujav1.CreateVolumeResponse) { r.Export.Portal = "" }), nil + }}, cfg: cfg, req: CreateRequest{VolumeID: "v", SizeBytes: 1}}, + {name: "lun", api: &fakeAPI{create: func(context.Context, *koujav1.CreateVolumeRequest) (*koujav1.CreateVolumeResponse, error) { + return mutateCreated(base, func(r *koujav1.CreateVolumeResponse) { r.Export.Lun = -1 }), nil + }}, cfg: cfg, req: CreateRequest{VolumeID: "v", SizeBytes: 1}}, + {name: "owner", api: &fakeAPI{create: func(context.Context, *koujav1.CreateVolumeRequest) (*koujav1.CreateVolumeResponse, error) { + return mutateCreated(base, func(r *koujav1.CreateVolumeResponse) { r.Volume.Owner = "" }), nil + }}, cfg: cfg, req: CreateRequest{VolumeID: "v", SizeBytes: 1}}, + {name: "bad portal", api: &fakeAPI{create: func(context.Context, *koujav1.CreateVolumeRequest) (*koujav1.CreateVolumeResponse, error) { + return mutateCreated(base, func(r *koujav1.CreateVolumeResponse) { r.Export.Portal = "127.0.0.1:3260" }), nil + }}, cfg: cfg, req: CreateRequest{VolumeID: "v", SizeBytes: 1}}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + provisioner := &Provisioner{API: tt.api} + if _, err := provisioner.Create(context.Background(), tt.cfg, tt.req); err == nil { + t.Fatal("Create() accepted invalid input") + } + }) + } + if _, err := (&Provisioner{API: &fakeAPI{}}).Create(context.Background(), cfg, CreateRequest{SizeBytes: 1}); err == nil { + t.Fatal("Create() accepted an empty volume ID") + } + if _, err := (&Provisioner{API: &fakeAPI{}}).Create(context.Background(), cfg, CreateRequest{VolumeID: "v"}); err == nil { + t.Fatal("Create() accepted an empty size") + } +} + +func mutateCreated(base *koujav1.CreateVolumeResponse, mutate func(*koujav1.CreateVolumeResponse)) *koujav1.CreateVolumeResponse { + value := &koujav1.CreateVolumeResponse{ + Volume: &koujav1.Volume{ + Owner: base.Volume.Owner, VolumeId: base.Volume.VolumeId, + ResourceUuid: base.Volume.ResourceUuid, ActualSizeBytes: base.Volume.ActualSizeBytes, + State: base.Volume.State, + }, + Export: &koujav1.Export{ + Protocol: base.Export.Protocol, TargetName: base.Export.TargetName, + Portal: base.Export.Portal, Lun: base.Export.Lun, Nsid: base.Export.Nsid, + State: base.Export.State, + }, + } + mutate(value) + return value +}