From 3bef5a9ad7144b24eea4985c445f2feb6f9a9cc4 Mon Sep 17 00:00:00 2001 From: Dan Fuhry Date: Mon, 19 Oct 2026 16:20:42 -0400 Subject: [PATCH] [sqpkg] add sqpkg-upload tool - relocate direct interactions with object storage into k8s/sqpkg package - add //cmd/sqpkg_upload --- cmd/sqpkg_upload/BUILD.bazel | 32 ++ cmd/sqpkg_upload/main.go | 359 ++++++++++++++++++++++ k8s/sqpkg/BUILD.bazel | 13 + k8s/sqpkg/client.go | 495 +++++++++++++++++++++++++++++++ k8s/sqpkg/csi_driver/BUILD.bazel | 3 +- k8s/sqpkg/csi_driver/node.go | 276 +++++++++++++++-- k8s/sqpkg/csi_driver/s3.go | 141 --------- 7 files changed, 1154 insertions(+), 165 deletions(-) create mode 100644 cmd/sqpkg_upload/BUILD.bazel create mode 100644 cmd/sqpkg_upload/main.go create mode 100644 k8s/sqpkg/BUILD.bazel create mode 100644 k8s/sqpkg/client.go delete mode 100644 k8s/sqpkg/csi_driver/s3.go diff --git a/cmd/sqpkg_upload/BUILD.bazel b/cmd/sqpkg_upload/BUILD.bazel new file mode 100644 index 0000000..e027bc0 --- /dev/null +++ b/cmd/sqpkg_upload/BUILD.bazel @@ -0,0 +1,32 @@ +load("//bazel:cross.bzl", "go_cross_binaries") +load("//bazel:go.bzl", "go_multi_binary", "go_multi_library") + +package(default_visibility = ["//visibility:public"]) + +go_multi_library( + name = "sqpkg_upload_lib", + srcs = ["main.go"], + importpath = "go.fuhry.dev/runtime/cmd/sqpkg_upload", + visibility = ["//visibility:public"], + deps = [ + "//constants", + "//k8s/sqpkg", + "//utils/context", + "//utils/hashset", + "//utils/log", + "@com_github_urfave_cli_altsrc_v3//:cli-altsrc", + "@com_github_urfave_cli_altsrc_v3//yaml", + "@com_github_urfave_cli_v3//:cli", + ], +) + +go_multi_binary( + name = "sqpkg_upload", + embed = [":sqpkg_upload_lib"], + visibility = ["//visibility:public"], +) + +go_cross_binaries( + base_target = "sqpkg_upload", + linux_arm = True, +) diff --git a/cmd/sqpkg_upload/main.go b/cmd/sqpkg_upload/main.go new file mode 100644 index 0000000..7a1a750 --- /dev/null +++ b/cmd/sqpkg_upload/main.go @@ -0,0 +1,359 @@ +package main + +import ( + "flag" + "fmt" + "os" + "path" + + altsrc "github.com/urfave/cli-altsrc/v3" + "github.com/urfave/cli-altsrc/v3/yaml" + "github.com/urfave/cli/v3" + + "go.fuhry.dev/runtime/constants" + "go.fuhry.dev/runtime/k8s/sqpkg" + "go.fuhry.dev/runtime/utils/context" + "go.fuhry.dev/runtime/utils/hashset" + "go.fuhry.dev/runtime/utils/log" +) + +type sqpkgUploadCommand struct { + sq sqpkg.SqpkgClient +} + +var logger log.Logger + +func (s *sqpkgUploadCommand) cmdUpload(ctx context.Context, c *cli.Command) error { + pkg := c.Value("package-name").(string) + sqfs := c.Value("source").(string) + var tags []string + if t := c.Value("tag"); t != nil { + tags = t.([]string) + } + + fp, err := os.OpenFile(sqfs, os.O_RDONLY, os.FileMode(0)) + if err != nil { + return fmt.Errorf("error opening sqfs package %q for reading: %v", sqfs, err) + } + defer fp.Close() + + logger.Infof("package name: %q", pkg) + logger.Infof("sqfs package: %q", sqfs) + logger.Infof("tags: %+v", tags) + sum, err := s.sq.UploadPackage(ctx, fp, pkg) + if err != nil { + return err + } + + for _, tag := range tags { + if err := s.sq.TagPackage(ctx, pkg, sum, tag); err == nil { + logger.Infof("wrote tag: %q", tag) + } else { + logger.Warningf("failed to tag package %q=%s@%s: %v", pkg, tag, sum, err) + } + } + + return nil +} + +func (s *sqpkgUploadCommand) cmdListPackages(ctx context.Context, c *cli.Command) error { + quiet := c.Value("quiet").(bool) + + for pkgInfo := range s.sq.ListPackages(ctx) { + if pkgInfo.Err != nil { + return pkgInfo.Err + } + + if quiet { + fmt.Println(pkgInfo.Name) + } else { + fmt.Printf( + "%s (pkgs: %d; tags: %d)\n", + pkgInfo.Name, + pkgInfo.PackageCount, + pkgInfo.TagCount, + ) + } + } + + return nil +} + +func (s *sqpkgUploadCommand) cmdListPackageSums(ctx context.Context, c *cli.Command) error { + pkg := c.Value("package-name").(string) + + for sum := range s.sq.ListSums(ctx, pkg) { + if sum.Err != nil { + return sum.Err + } + + fmt.Println(sum.Sum) + } + + return nil +} + +func (s *sqpkgUploadCommand) cmdListTags(ctx context.Context, c *cli.Command) error { + quiet := c.Value("quiet").(bool) + pkg := c.Value("package-name").(string) + + for tag := range s.sq.ListTags(ctx, pkg) { + if tag.Err != nil { + return tag.Err + } + + if quiet { + fmt.Println(tag.Tag) + } else { + fmt.Printf("%s:%s\n", tag.Tag, tag.Sum) + } + } + + return nil +} + +func (s *sqpkgUploadCommand) cmdPruneOrphans(ctx context.Context, c *cli.Command) error { + pkg := c.Value("package-name").(string) + dry := c.Value("dry-run").(bool) + sums := hashset.NewHashSet[string]() + + for tag := range s.sq.ListTags(ctx, pkg) { + if tag.Err != nil { + return tag.Err + } + + sums.Add(tag.Sum) + } + + for sum := range s.sq.ListSums(ctx, pkg) { + if sum.Err != nil { + return sum.Err + } + + if !sums.Contains(sum.Sum) { + if dry { + logger.Noticef("package %q: sum %s would be pruned", pkg, sum.Sum) + } else { + if err := s.sq.DeletePackage(ctx, pkg, sum.Sum); err == nil { + logger.Noticef("package %q: pruned version with sum %s", pkg, sum.Sum) + } else { + logger.Errorf("package %q: failed to prune version with sum %s: %v", pkg, sum.Sum, err) + } + } + } + } + + return nil +} + +func (s *sqpkgUploadCommand) cmdDownload(ctx context.Context, c *cli.Command) error { + pkg := c.StringArg("package") + ver := c.StringArg("vers") + dst := c.Value("dest").(string) + + if sqpkg.ValidTag(ver) { + var err error + ver, err = s.sq.ResolveTag(ctx, pkg, ver) + if err != nil { + return err + } + } + + if !sqpkg.ValidShaSum(ver) { + return fmt.Errorf("invalid sha or tag name: %q", ver) + } + + if dst == "" { + dst = fmt.Sprintf("%s-%s.sqfs", pkg, ver) + } + + fp, err := os.OpenFile(dst, os.O_CREATE|os.O_WRONLY, os.FileMode(0o644)) + if err != nil { + return err + } + defer fp.Close() + + err = s.sq.DownloadPackage(ctx, fp, pkg, ver) + if err != nil { + return err + } + + logger.Infof("package written to: %s", dst) + + return nil +} + +func yamlSrc(keyName string) cli.ValueSourceChain { + c := cli.NewValueSourceChain() + if home := os.Getenv("HOME"); home != "" { + settingsPath := path.Join(home, ".config", constants.OrgSlug, "settings.yaml") + c.Chain = append(c.Chain, yaml.YAML(keyName, altsrc.StringSourcer(settingsPath))) + } + + return c +} + +func main() { + ctx, cancel := context.Interruptible() + defer cancel() + + cmd := &sqpkgUploadCommand{} + + type flagCollection struct { + pkg, tag, src, dst, dry, quiet cli.Flag + } + flags := flagCollection{ + pkg: &cli.StringFlag{ + Name: "package-name", + Aliases: []string{"p"}, + Usage: "pkg", + OnlyOnce: true, + Required: true, + }, + tag: &cli.StringSliceFlag{ + Name: "tag", + Aliases: []string{"t"}, + Usage: "tag_name", + OnlyOnce: false, + Required: false, + }, + src: &cli.StringFlag{ + Name: "source", + Aliases: []string{"s"}, + Usage: "package.sqfs", + OnlyOnce: true, + Required: true, + TakesFile: true, + }, + dst: &cli.StringFlag{ + Name: "dest", + Aliases: []string{"d"}, + Usage: "package.sqfs", + OnlyOnce: true, + TakesFile: true, + }, + dry: &cli.BoolFlag{ + Name: "dry-run", + Aliases: []string{"n"}, + Usage: "only preview what would be done", + Value: true, + }, + quiet: &cli.BoolFlag{ + Name: "quiet", + Aliases: []string{"q"}, + Usage: "minimal output for use with scripts, etc.", + Value: false, + }, + } + + app := &cli.Command{ + Name: constants.OrgSlug + "-sqpkg-upload", + Description: "manage sqpkgs in the package store", + Version: constants.Version, + AllowExtFlags: true, + Commands: []*cli.Command{ + { + Name: "upload", + Description: "upload a package", + Flags: []cli.Flag{ + flags.pkg, + flags.tag, + flags.src, + }, + Action: cmd.cmdUpload, + }, + { + Name: "list-packages", + Aliases: []string{"lp"}, + Description: "list the packages in the store", + Flags: []cli.Flag{flags.quiet}, + Action: cmd.cmdListPackages, + }, + { + Name: "list-sums", + Aliases: []string{"ls"}, + Description: "list all images for a package, tagged or not", + Action: cmd.cmdListPackageSums, + Flags: []cli.Flag{flags.pkg}, + }, + { + Name: "list-tags", + Aliases: []string{"lt"}, + Description: "list tagged versions for a package", + Action: cmd.cmdListTags, + Flags: []cli.Flag{flags.pkg, flags.quiet}, + }, + { + Name: "prune-orphans", + Aliases: []string{"p", "prune"}, + Description: "delete all versions of a package that aren't tagged", + Action: cmd.cmdPruneOrphans, + Flags: []cli.Flag{flags.pkg, flags.dry}, + }, + { + Name: "download", + Aliases: []string{"dl"}, + Description: "download a package", + Action: cmd.cmdDownload, + Flags: []cli.Flag{flags.dst}, + Arguments: []cli.Argument{ + &cli.StringArg{ + Name: "package", + UsageText: "package-name", + }, + &cli.StringArg{ + Name: "vers", + UsageText: "sum-or-tag", + }, + }, + }, + }, + Flags: []cli.Flag{ + &cli.StringFlag{ + Name: "s3.endpoint", + Category: "Object storage settings:", + Usage: "s3-endpoint-hostname", + Sources: yamlSrc("packagestore.endpoint"), + }, + &cli.StringFlag{ + Name: "s3.bucket", + Category: "Object storage settings:", + Usage: "s3-bucket-name", + Sources: yamlSrc("packagestore.bucket"), + }, + &cli.StringFlag{ + Name: "s3.access-key", + Category: "Object storage settings:", + Usage: "", + Sources: yamlSrc("packagestore.credentials.accessKey"), + }, + &cli.StringFlag{ + Name: "s3.secret-key", + Category: "Object storage settings:", + Usage: "", + Sources: yamlSrc("packagestore.credentials.secretKey"), + }, + }, + Before: func(ctx context.Context, c *cli.Command) (context.Context, error) { + endpoint := c.Value("s3.endpoint").(string) + bucket := c.Value("s3.bucket").(string) + accessKey := c.Value("s3.access-key").(string) + secretKey := c.Value("s3.secret-key").(string) + config := sqpkg.NewConfigWithAccessKeys(endpoint, bucket, accessKey, secretKey) + cmd.sq = sqpkg.NewClient(config) + + return ctx, nil + }, + } + flag.CommandLine.VisitAll(func(f *flag.Flag) { + app.Flags = append(app.Flags, &cli.GenericFlag{ + Name: f.Name, + Usage: f.Usage, + Value: f.Value.(cli.Value), + }) + }) + logger = log.Default() + if err := app.Run(ctx, os.Args); err != nil { + logger.Fatal(err) + } +} diff --git a/k8s/sqpkg/BUILD.bazel b/k8s/sqpkg/BUILD.bazel new file mode 100644 index 0000000..e6a6e71 --- /dev/null +++ b/k8s/sqpkg/BUILD.bazel @@ -0,0 +1,13 @@ +load("//bazel:go.bzl", "go_multi_library") + +go_multi_library( + name = "sqpkg", + srcs = ["client.go"], + importpath = "go.fuhry.dev/runtime/k8s/sqpkg", + visibility = ["//visibility:public"], + deps = [ + "//utils/log", + "@com_github_minio_minio_go_v7//:minio-go", + "@com_github_minio_minio_go_v7//pkg/credentials", + ], +) diff --git a/k8s/sqpkg/client.go b/k8s/sqpkg/client.go new file mode 100644 index 0000000..5f2e472 --- /dev/null +++ b/k8s/sqpkg/client.go @@ -0,0 +1,495 @@ +package sqpkg + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/hex" + "errors" + "fmt" + "io" + "os" + "path" + "regexp" + "strings" + "sync" + + "github.com/minio/minio-go/v7" + "github.com/minio/minio-go/v7/pkg/credentials" + + "go.fuhry.dev/runtime/utils/log" +) + +type PackageInfo struct { + Err error + Name string + PackageCount, TagCount uint +} + +type SumListEntry struct { + Err error + Sum string +} + +type TagListEntry struct { + Err error + Tag, Sum string +} + +type ReadSeekerStat interface { + io.ReadSeeker + + Stat() (os.FileInfo, error) +} + +type TagMap = map[string]string + +type Config interface { + Endpoint() string + BucketName() string + Credentials() *credentials.Credentials +} + +type SqpkgClient interface { + Config + + ListPackages(ctx context.Context) <-chan PackageInfo + ListSums(ctx context.Context, pkg string) <-chan SumListEntry + ListTags(ctx context.Context, pkg string) <-chan TagListEntry + ResolveTag(ctx context.Context, pkg, tag string) (string, error) + DownloadPackage(ctx context.Context, w io.Writer, pkg, sum string) error + UploadPackage(ctx context.Context, r ReadSeekerStat, pkg string) (string, error) + DeletePackage(ctx context.Context, pkg, sum string) error + TagPackage(ctx context.Context, pkg, sum, tag string) error + UntagPackage(ctx context.Context, pkg, tag string) error +} + +var ( + validPackageName = regexp.MustCompile(`^[a-z0-9]([a-z0-9_-]*[a-z0-9])?$`) + validShaSum = regexp.MustCompile(`^[0-9a-f]{64}$`) + validTag = regexp.MustCompile(`^[a-z0-9]([a-z0-9\.+~_-]*[a-z0-9]$)?`) + validPkgPath = regexp.MustCompile(`^[0-9a-f]{2}/[0-9a-f]{2}/[0-9a-f]{64}$`) +) + +var ( + ValidPackageName = validPackageName.MatchString + ValidShaSum = validShaSum.MatchString + ValidTag = validTag.MatchString + ValidPkgPath = validPkgPath.MatchString +) + +func ObjectKey(pkg, sum string) string { + return path.Join(pkg, sum[:2], sum[2:4], sum) +} + +func TagKey(pkg, tag string) string { + return path.Join(pkg, "tags", tag) +} + +type config struct { + endpoint, bucket, accessKey, secretKey string +} + +var _ Config = &config{} + +func (c *config) Endpoint() string { + return c.endpoint +} + +func (c *config) BucketName() string { + return c.bucket +} + +func (c *config) Credentials() *credentials.Credentials { + if c.accessKey != "" && c.secretKey != "" { + return credentials.NewStaticV4(c.accessKey, c.secretKey, "") + } + + return credentials.NewEnvAWS() +} + +func NewConfig(endpoint, bucket string) Config { + return &config{ + endpoint: endpoint, + bucket: bucket, + } +} + +func NewConfigWithAccessKeys(endpoint, bucket, accessKey, secretKey string) Config { + return &config{ + endpoint: endpoint, + bucket: bucket, + accessKey: accessKey, + secretKey: secretKey, + } +} + +type client struct { + Config + + logger log.Logger + client *minio.Client + clientMu sync.Mutex +} + +var _ SqpkgClient = &client{} + +func NewClient(config Config) SqpkgClient { + return &client{ + Config: config, + logger: log.Default().WithPrefix("sqpkg.SqpkgClient"), + } +} + +func isPackageDir(ctx context.Context, client *minio.Client, bucket, dir string) (*PackageInfo, error) { + if !ValidPackageName(strings.TrimSuffix(dir, "/")) { + return nil, nil + } + + i := &PackageInfo{} + opts := minio.ListObjectsOptions{ + Prefix: dir, + Recursive: true, + } + ctx, cancel := context.WithCancel(ctx) + defer cancel() + + for entry := range client.ListObjects(ctx, bucket, opts) { + if entry.Err != nil { + if errors.Is(entry.Err, context.Canceled) { + return nil, nil + } + + return nil, entry.Err + } + + subpath := entry.Key[len(dir):] + if strings.HasPrefix(subpath, "tags/") && ValidTag(subpath[5:]) { + i.TagCount++ + } else if ValidPkgPath(subpath) { + i.PackageCount++ + } else { + cancel() + } + } + + return i, nil +} + +func (c *client) ListPackages(ctx context.Context) <-chan PackageInfo { + ch := make(chan PackageInfo) + + go func() { + defer close(ch) + + client, err := c.s3Client() + if err != nil { + ch <- PackageInfo{Err: err} + return + } + + objects := client.ListObjects(ctx, c.BucketName(), minio.ListObjectsOptions{Recursive: false}) + for object := range objects { + if object.Err != nil { + ch <- PackageInfo{Err: err} + return + } + + pkgInfo, err := isPackageDir(ctx, client, c.BucketName(), object.Key) + if err != nil { + c.logger.Warningf("error checking whether key %q is a package: %v", object.Key, err) + continue + } + + if pkgInfo != nil { + pkgInfo.Name = object.Key[:len(object.Key)-1] + + ch <- *pkgInfo + } + } + }() + + return ch +} + +func (c *client) ListSums(ctx context.Context, pkg string) <-chan SumListEntry { + ch := make(chan SumListEntry) + + go func() { + defer close(ch) + + if !ValidPackageName(pkg) { + ch <- SumListEntry{Err: fmt.Errorf("invalid package name: %q", pkg)} + return + } + + client, err := c.s3Client() + if err != nil { + ch <- SumListEntry{Err: err} + return + } + + opts := minio.ListObjectsOptions{ + Recursive: true, + Prefix: pkg + "/", + } + + for entry := range client.ListObjects(ctx, c.BucketName(), opts) { + if entry.Err != nil { + ch <- SumListEntry{Err: err} + return + } + + entryName := strings.TrimPrefix(entry.Key, opts.Prefix) + if ValidPkgPath(entryName) { + ch <- SumListEntry{Sum: path.Base(entryName)} + } + } + }() + + return ch +} + +func (c *client) ListTags(ctx context.Context, pkg string) <-chan TagListEntry { + ch := make(chan TagListEntry) + + go func() { + defer close(ch) + + if !ValidPackageName(pkg) { + ch <- TagListEntry{Err: fmt.Errorf("invalid package name: %q")} + return + } + + client, err := c.s3Client() + if err != nil { + ch <- TagListEntry{Err: err} + return + } + + opts := minio.ListObjectsOptions{ + Recursive: false, + Prefix: fmt.Sprintf("%s/tags/", pkg), + } + + for entry := range client.ListObjects(ctx, c.BucketName(), opts) { + if entry.Err != nil { + if errors.Is(entry.Err, context.Canceled) { + return + } + + ch <- TagListEntry{Err: err} + return + } + + tagName := entry.Key[len(opts.Prefix):] + reader, err := client.GetObject(ctx, c.BucketName(), entry.Key, minio.GetObjectOptions{}) + if err == nil { + tagBuf := bytes.NewBuffer(nil) + io.Copy(tagBuf, reader) + reader.Close() + sha := tagBuf.String() + + if ValidShaSum(sha) { + ch <- TagListEntry{ + Tag: tagName, + Sum: sha, + } + } + } else { + c.logger.Warningf("failed to read tag %q: %v", tagName, err) + } + } + }() + + return ch +} + +func (c *client) ResolveTag(ctx context.Context, pkg, tag string) (string, error) { + if !ValidTag(tag) { + return "", fmt.Errorf("invalid tag name: %q", tag) + } + + client, err := c.s3Client() + if err != nil { + return "", err + } + tagPath := path.Join(pkg, "tags", tag) + buf := bytes.NewBuffer(nil) + + reader, err := client.GetObject(ctx, c.BucketName(), tagPath, minio.GetObjectOptions{}) + if err != nil { + return "", err + } + io.Copy(buf, reader) + reader.Close() + + sha := strings.TrimSpace(buf.String()) + + if !ValidShaSum(sha) { + return "", fmt.Errorf("tag contains invalid shasum: %q", sha) + } + + return sha, nil +} + +func (c *client) DownloadPackage(ctx context.Context, w io.Writer, pkg, sum string) error { + if !ValidPackageName(pkg) { + return fmt.Errorf("invalid package name: %q", pkg) + } + if !ValidShaSum(sum) { + return fmt.Errorf("invalid shasum: %q", sum) + } + + client, err := c.s3Client() + if err != nil { + return err + } + + reader, err := client.GetObject(ctx, c.BucketName(), ObjectKey(pkg, sum), minio.GetObjectOptions{}) + if err != nil { + return err + } + defer reader.Close() + + _, err = io.Copy(w, reader) + if errors.Is(err, io.EOF) { + return nil + } + return err +} + +func (c *client) UploadPackage(ctx context.Context, r ReadSeekerStat, pkg string) (string, error) { + if !ValidPackageName(pkg) { + return "", fmt.Errorf("invalid package name: %q", pkg) + } + + client, err := c.s3Client() + if err != nil { + return "", err + } + + var sum string + stat, err := r.Stat() + if err != nil { + return "", err + } + sha := sha256.New() + io.Copy(sha, r) + + sum = hex.EncodeToString(sha.Sum(nil)) + key := ObjectKey(pkg, sum) + _, err = client.StatObject(ctx, c.BucketName(), key, minio.StatObjectOptions{}) + if err != nil { + if errResp := minio.ToErrorResponse(err); err != nil && errResp.Code != "NoSuchKey" { + return "", fmt.Errorf("HeadObject failed: %v: %v", errResp.Code, errResp.Message) + } + + r.Seek(0, 0) + + uploadInfo, err := client.PutObject(ctx, c.BucketName(), key, r, stat.Size(), minio.PutObjectOptions{}) + if err != nil { + return "", fmt.Errorf("PutObject(%q, %q) failed: %v", c.BucketName(), key, err) + } + c.logger.Noticef("%d bytes uploaded to key %q", uploadInfo.Size, uploadInfo.Key) + } else { + c.logger.Noticef("package %q with checksum %q already present, not uploading", pkg, sum) + } + + return sum, nil +} + +func (c *client) TagPackage(ctx context.Context, pkg, sum, tag string) error { + if !ValidPackageName(pkg) { + return fmt.Errorf("invalid package name: %q", pkg) + } + if !ValidTag(tag) { + return fmt.Errorf("tag %q is invalid", tag) + } + + client, err := c.s3Client() + if err != nil { + return err + } + + tagKey := path.Join(pkg, "tags", tag) + sumBuf := bytes.NewBufferString(sum) + _, err = client.PutObject(ctx, c.BucketName(), tagKey, sumBuf, int64(sumBuf.Len()), minio.PutObjectOptions{}) + if err != nil { + return fmt.Errorf("failed to write tag %q: %v", tag, err) + } + + return nil +} + +func (c *client) UntagPackage(ctx context.Context, pkg, tag string) error { + if !ValidPackageName(pkg) { + return fmt.Errorf("invalid package name: %q", pkg) + } + if !ValidTag(tag) { + return fmt.Errorf("tag %q is invalid", tag) + } + + client, err := c.s3Client() + if err != nil { + return err + } + + tagKey := TagKey(pkg, tag) + return client.RemoveObject(ctx, c.BucketName(), tagKey, minio.RemoveObjectOptions{}) +} + +func (c *client) DeletePackage(ctx context.Context, pkg, sum string) error { + if !ValidPackageName(pkg) { + return fmt.Errorf("invalid package name: %q", pkg) + } + if !ValidShaSum(sum) { + return fmt.Errorf("shasum %q is invalid", sum) + } + + for tag := range c.ListTags(ctx, pkg) { + if tag.Err != nil { + return tag.Err + } + + if tag.Sum == sum { + return fmt.Errorf("shasum %q is referenced by tag %q, please untag first", sum, tag.Tag) + } + } + + client, err := c.s3Client() + if err != nil { + return err + } + + packageKey := ObjectKey(pkg, sum) + return client.RemoveObject(ctx, c.BucketName(), packageKey, minio.RemoveObjectOptions{}) +} + +func (c *client) s3Client() (*minio.Client, error) { + c.clientMu.Lock() + defer c.clientMu.Unlock() + + if c.client == nil { + creds := c.Credentials() + + cc := &credentials.CredContext{ + Endpoint: c.Endpoint(), + } + + if _, err := creds.GetWithContext(cc); err != nil { + return nil, errors.New("cannot load credentials") + } + + minioClient, err := minio.New(c.Endpoint(), &minio.Options{ + Creds: creds, + Secure: true, + }) + if err != nil { + return nil, err + } + + c.client = minioClient + } + return c.client, nil +} diff --git a/k8s/sqpkg/csi_driver/BUILD.bazel b/k8s/sqpkg/csi_driver/BUILD.bazel index b5400ca..a420fea 100644 --- a/k8s/sqpkg/csi_driver/BUILD.bazel +++ b/k8s/sqpkg/csi_driver/BUILD.bazel @@ -6,16 +6,15 @@ go_multi_library( "controller.go", "identity.go", "node.go", - "s3.go", ], importpath = "go.fuhry.dev/runtime/k8s/sqpkg/csi_driver", visibility = ["//visibility:public"], deps = [ "//constants", + "//k8s/sqpkg", "//utils/context", "//utils/option", "@com_github_container_storage_interface_spec//lib/go/csi", - "@com_github_minio_minio_go_v7//:minio-go", "@com_github_minio_minio_go_v7//pkg/credentials", "@io_k8s_klog_v2//:klog", "@org_golang_google_grpc//codes", diff --git a/k8s/sqpkg/csi_driver/node.go b/k8s/sqpkg/csi_driver/node.go index 32c10c9..958e0d6 100644 --- a/k8s/sqpkg/csi_driver/node.go +++ b/k8s/sqpkg/csi_driver/node.go @@ -1,11 +1,20 @@ package csi_driver import ( + "bytes" + "crypto/sha256" + "encoding/hex" + "errors" + "fmt" + "hash" + "io" "os" "os/exec" "path" + "strconv" "strings" "sync" + "syscall" "k8s.io/klog/v2" @@ -14,21 +23,20 @@ import ( "google.golang.org/protobuf/encoding/prototext" "go.fuhry.dev/runtime/constants" + "go.fuhry.dev/runtime/k8s/sqpkg" "go.fuhry.dev/runtime/utils/context" "go.fuhry.dev/runtime/utils/option" "github.com/container-storage-interface/spec/lib/go/csi" + "github.com/minio/minio-go/v7/pkg/credentials" ) var ( kPackageName = "sqpkg.csi.k8s." + constants.RootDomain + "/package" kShaSum = "sqpkg.csi.k8s." + constants.RootDomain + "/build-id" + kTagName = "sqpkg.csi.k8s." + constants.RootDomain + "/tag" ) -type s3Params struct { - endpointUrl, bucketName, accessKey, secretKey string -} - type imageRef struct { pkg, sha string } @@ -43,7 +51,9 @@ type NodeServer struct { refCounts map[imageRef]uint refCountsLock sync.RWMutex - *s3Params + sqpkg.Config + client sqpkg.SqpkgClient + clientMu sync.Mutex } type NodeServerOption = option.Option[*NodeServer] @@ -51,6 +61,16 @@ func (r imageRef) imagePath() string { return path.Join(r.pkg, r.sha[:2], r.sha[2:4], r.sha) } +func (ns *NodeServer) getClient() sqpkg.SqpkgClient { + ns.clientMu.Lock() + defer ns.clientMu.Unlock() + + if ns.client == nil { + ns.client = sqpkg.NewClient(ns.Config) + } + return ns.client +} + func (ns *NodeServer) NodeGetInfo(ctx context.Context, req *csi.NodeGetInfoRequest) (*csi.NodeGetInfoResponse, error) { return &csi.NodeGetInfoResponse{ NodeId: ns.nodeID, @@ -111,33 +131,55 @@ func (ns *NodeServer) NodePublishVolume(ctx context.Context, req *csi.NodePublis return &csi.NodePublishVolumeResponse{}, nil } - var pkg, shasum string + var pkg, sha, tag string for ctxKey, value := range req.GetVolumeContext() { switch ctxKey { case kPackageName: pkg = value case kShaSum: - shasum = value + sha = value + case kTagName: + tag = value } } - if pkg == "" || shasum == "" { + if pkg == "" || (sha == "" && tag == "") { + return nil, status.Errorf( + codes.InvalidArgument, + "missing one or more keys: %q, (%q || %q)", kPackageName, kShaSum, kTagName, + ) + } else if sha != "" && tag != "" { return nil, status.Errorf( codes.InvalidArgument, - "missing one or more keys: %q, %q", kPackageName, kShaSum) + "%q and %q are mutually exclusive", kShaSum, kTagName, + ) + } else if tag != "" { + var err error + sha, err = ns.getClient().ResolveTag(ctx, pkg, tag) + if err != nil { + return nil, status.Errorf( + codes.NotFound, + "tag not found for package %q: %q", pkg, tag, + ) + } } - sqfs, err := ns.maybeGetPackage(ctx, pkg, shasum) + sqfs, err := ns.maybeGetPackage(ctx, pkg, sha) if err != nil { return nil, status.Errorf( codes.Internal, - "error loading package %s/%s: %v", pkg, shasum, err) + "error loading package %s/%s: %v", pkg, sha, err, + ) } if err := os.MkdirAll(targetPath, 0o755); err != nil { return nil, status.Errorf(codes.Internal, "Failed to create target directory: %v", err) } + if err := ensureNextLoopAvailable(); err != nil { + return nil, status.Errorf(codes.Internal, "failed to ensure the next loop device is available: %v", err) + } + mountArgs := []string{ "/usr/bin/mount", "-t", "squashfs", @@ -152,10 +194,11 @@ func (ns *NodeServer) NodePublishVolume(ctx context.Context, req *csi.NodePublis if err := cmd.Run(); err != nil { return nil, status.Errorf( codes.Internal, - "error running command %q: %v", strings.Join(mountArgs, " "), err) + "error running command %q: %v", strings.Join(mountArgs, " "), err, + ) } - ns.increment(pkg, shasum, targetPath) + ns.increment(pkg, sha, targetPath) return &csi.NodePublishVolumeResponse{}, nil } @@ -174,7 +217,8 @@ func (ns *NodeServer) NodeUnpublishVolume(ctx context.Context, req *csi.NodeUnpu // handle any errors return nil, status.Errorf( codes.Internal, - "failed to unmount volume from %q: %v", targetPath, err) + "failed to unmount volume from %q: %v", targetPath, err, + ) } os.RemoveAll(targetPath) @@ -211,24 +255,40 @@ func WithCacheDir(cacheDir string) NodeServerOption { func WithS3Params(endpointUrl, bucketName string) NodeServerOption { return option.NewOption(func(s *NodeServer) error { - if s.s3Params == nil { - s.s3Params = &s3Params{} + var creds *credentials.Credentials + if s.Config != nil { + creds = s.Config.Credentials() } - s.s3Params.endpointUrl = endpointUrl - s.s3Params.bucketName = bucketName + if creds != nil { + cc := &credentials.CredContext{Endpoint: endpointUrl} + credValue, err := creds.GetWithContext(cc) + if err != nil { + return err + } + + s.Config = sqpkg.NewConfigWithAccessKeys( + endpointUrl, + bucketName, + credValue.AccessKeyID, + credValue.SecretAccessKey, + ) + return nil + } + s.Config = sqpkg.NewConfig(endpointUrl, bucketName) return nil }) } func WithS3Credentials(accessKey, secretKey string) NodeServerOption { return option.NewOption(func(s *NodeServer) error { - if s.s3Params == nil { - s.s3Params = &s3Params{} + var endpointUrl, bucketName string + if s.Config != nil { + endpointUrl = s.Config.Endpoint() + bucketName = s.Config.BucketName() } - s.s3Params.accessKey = accessKey - s.s3Params.secretKey = secretKey + s.Config = sqpkg.NewConfigWithAccessKeys(endpointUrl, bucketName, accessKey, secretKey) return nil }) } @@ -246,3 +306,175 @@ func NewNodeServer(nodeID string, options ...NodeServerOption) csi.NodeServer { } return s } + +func (ns *NodeServer) maybeGetPackage(ctx context.Context, packageName, shasum string) (string, error) { + cachePath := path.Join( + ns.cacheDir, + packageName, + shasum[:2], + shasum[2:4], + shasum, + ) + klog.V(1).Infof("maybeGetPackage: cachePath: %q", cachePath) + + if !sqpkg.ValidShaSum(shasum) { + return "", fmt.Errorf("invalid shasum: %q", shasum) + } + + cacheDir := path.Dir(cachePath) + if err := os.MkdirAll(cacheDir, os.FileMode(0o755)); err != nil && !os.IsExist(err) { + return "", err + } + + shaBytes, err := hex.DecodeString(shasum) + if err != nil { + return "", err + } + + fp, err := os.OpenFile(cachePath, os.O_RDWR|os.O_CREATE, os.FileMode(0o644)) + if err != nil { + return "", err + } + defer fp.Close() + stat, err := fp.Stat() + if err != nil { + return "", err + } + if stat.Size() == 0 { + klog.V(1).Infof(" -> cachePath did not previously exist, attempting to download") + if err := ns.getClient().DownloadPackage(ctx, fp, packageName, shasum); err != nil { + os.Remove(cachePath) + klog.Errorf("failed to download package %q sum %q to %s: %v", packageName, shasum, cachePath, err) + return "", err + } + fp.Sync() + fp.Seek(0, 0) + } + var hasher hash.Hash = sha256.New() + io.Copy(hasher, fp) + hashBuf := make([]byte, 0) + hashBuf = hasher.Sum(hashBuf) + + if !bytes.Equal(hashBuf, shaBytes) { + return "", fmt.Errorf( + "package at path %q does not match expected checksum:\n expected: %s\n actual: %s", + cachePath, shasum, hex.EncodeToString(hashBuf), + ) + } + + return cachePath, nil +} + +func isSquashFsMounted(mountPath string) bool { + mountsBytes, err := os.ReadFile("/proc/mounts") + if err != nil { + return false + } + + mounts := strings.Split(strings.TrimSpace(string(mountsBytes)), "\n") + for _, line := range mounts { + parts := strings.SplitN(line, " ", 6) + if len(parts) >= 3 { + if strings.HasPrefix(parts[0], "/dev/loop") && + parts[1] == mountPath && + parts[2] == "squashfs" { + return true + } + } + } + + return false +} + +func makedev(major, minor uint) uint64 { + return (uint64(major) & 0xfff << 8) | + (uint64(major) & 0xfffff000 << 32) | + (uint64(minor) & 0xff) | + (uint64(minor) & 0xffffff00 << 12) +} + +func ensureNextLoopAvailable() error { + minor, err := getNextLoopDevice() + if err != nil { + return err + } + + devPath := fmt.Sprintf("/dev/loop%d", minor) + stat, err := os.Stat(devPath) + if err != nil { + if os.IsNotExist(err) { + major, err := getLoopMajorNumber() + if err != nil { + return err + } + + klog.Infof("creating loop device %q with major %d/minor %d", devPath, major, minor) + + devno := makedev(major, minor) + return syscall.Mknod(devPath, syscall.S_IFBLK|0o600, int(devno)) + } + return err + } + + posixStat, ok := stat.Sys().(*syscall.Stat_t) + if !ok { + return fmt.Errorf("stat.Sys() returned %T, not *syscall.Stat_t", stat.Sys()) + } + + if syscall.S_IFBLK != (posixStat.Mode & syscall.S_IFBLK) { + return fmt.Errorf("%s exists but is not a block device", devPath) + } + + return nil +} + +func getNextLoopDevice() (uint, error) { + cmd := exec.Command("losetup", "-f") + stdout := &bytes.Buffer{} + cmd.Stdout = stdout + + if err := cmd.Run(); err != nil { + return 0, cmd.Err + } + + if !cmd.ProcessState.Success() { + return 0, errors.New("command 'losetup -f' exited with error") + } + + out := strings.TrimPrefix(strings.TrimSpace(stdout.String()), "/dev/loop") + out = strings.TrimSuffix(out, " (lost)") + if n, err := strconv.Atoi(out); err == nil { + return uint(n), nil + } else { + return 0, err + } +} + +func getLoopMajorNumber() (uint, error) { + devices, err := os.ReadFile("/proc/devices") + if err != nil { + return 0, err + } + + lines := strings.Split(string(devices), "\n") + foundLines := false + for _, line := range lines { + if foundLines { + major, driver, ok := strings.Cut(strings.TrimSpace(line), " ") + if ok && driver == "loop" { + majorInt, err := strconv.Atoi(major) + if err == nil { + return uint(majorInt), nil + } + + return 0, fmt.Errorf("failed to parse major number in line %q: %v", line, err) + } + } else { + if line == "Block devices:" { + foundLines = true + } + } + } + + return 0, errors.New("loop module not loaded or not found in /proc/devices") +} diff --git a/k8s/sqpkg/csi_driver/s3.go b/k8s/sqpkg/csi_driver/s3.go deleted file mode 100644 index f1bfbcc..0000000 --- a/k8s/sqpkg/csi_driver/s3.go +++ /dev/null @@ -1,141 +0,0 @@ -package csi_driver - -import ( - "bytes" - "context" - "crypto/sha256" - "encoding/hex" - "errors" - "fmt" - "hash" - "io" - "os" - "path" - "regexp" - "strings" - - "github.com/minio/minio-go/v7" - "github.com/minio/minio-go/v7/pkg/credentials" - "k8s.io/klog/v2" -) - -var ( - ValidPackageName = regexp.MustCompile(`^[a-z0-9]([a-z0-9_-]*[a-z0-9])?$`) - ValidShaSum = regexp.MustCompile(`^[0-9a-f]{64}$`) -) - -func (ns *NodeServer) maybeGetPackage(ctx context.Context, packageName, shasum string) (string, error) { - if !ValidPackageName.MatchString(packageName) { - return "", fmt.Errorf("packageName is not valid: %q", packageName) - } - - if !ValidShaSum.MatchString(shasum) { - return "", fmt.Errorf("shasum is not valid: %q", shasum) - } - - shaBytes, err := hex.DecodeString(shasum) - if err != nil { - return "", err - } - - cachePath := path.Join( - ns.cacheDir, - packageName, - shasum[:2], - shasum[2:4], - shasum, - ) - klog.V(1).Infof("maybeGetPackage: cachePath: %q", cachePath) - - fp, err := os.OpenFile(cachePath, os.O_RDONLY, os.FileMode(0)) - if err != nil && os.IsNotExist(err) { - klog.V(1).Infof(" -> cachePath does not exist, attempting to download") - if ns.s3Params == nil { - return "", errors.New("s3Params not populated") - } - - var creds *credentials.Credentials - - if ns.s3Params.accessKey != "" && ns.s3Params.secretKey != "" { - creds = credentials.NewStaticV4(ns.s3Params.accessKey, ns.s3Params.secretKey, "") - } else { - creds = credentials.NewEnvAWS() - } - - cc := &credentials.CredContext{ - Endpoint: ns.s3Params.endpointUrl, - } - - if _, err := creds.GetWithContext(cc); err != nil { - return "", errors.New("cannot load credentials") - } - - client, err := minio.New(ns.s3Params.endpointUrl, &minio.Options{ - Creds: creds, - Secure: true, - }) - if err != nil { - return "", err - } - - err = os.MkdirAll(path.Dir(cachePath), os.FileMode(0o700)) - if err != nil && !os.IsExist(err) { - return "", err - } - - packagePath := path.Join(packageName, shasum[:2], shasum[2:4], shasum) - - err = client.FGetObject( - ctx, - ns.s3Params.bucketName, - packagePath, - cachePath, - minio.GetObjectOptions{}) - if err != nil { - return "", err - } - } else if err != nil { - return "", err - } else if fp != nil { - fp.Close() - } - - fp, err = os.OpenFile(cachePath, os.O_RDONLY, os.FileMode(0o600)) - if err != nil { - return "", err - } - defer fp.Close() - var hasher hash.Hash = sha256.New() - io.Copy(hasher, fp) - hashBuf := make([]byte, 0) - hashBuf = hasher.Sum(hashBuf) - - if !bytes.Equal(hashBuf, shaBytes) { - return "", fmt.Errorf( - "package at path %q does not match expected checksum:\n expected: %s\n actual: %s", - cachePath, shasum, hex.EncodeToString(hashBuf)) - } - - return cachePath, nil -} - -func isSquashFsMounted(mountPath string) bool { - mountsBytes, err := os.ReadFile("/proc/mounts") - if err != nil { - return false - } - - mounts := strings.Split(strings.TrimSpace(string(mountsBytes)), "\n") - for _, line := range mounts { - parts := strings.SplitN(line, " ", 6) - if len(parts) >= 3 { - if strings.HasPrefix(parts[0], "/dev/loop") && - parts[1] == mountPath && - parts[2] == "squashfs" { - return true - } - } - } - - return false -} -- 2.52.0