--- /dev/null
+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,
+)
--- /dev/null
+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)
+ }
+}
--- /dev/null
+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",
+ ],
+)
--- /dev/null
+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
+}
"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",
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"
"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
}
refCounts map[imageRef]uint
refCountsLock sync.RWMutex
- *s3Params
+ sqpkg.Config
+ client sqpkg.SqpkgClient
+ clientMu sync.Mutex
}
type NodeServerOption = option.Option[*NodeServer]
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,
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",
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
}
// 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)
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
})
}
}
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")
+}
+++ /dev/null
-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
-}
projects / runtime.git / commitdiff