ziggi/mstore
1
0
Fork 0
Code Pull Requests Actions Packages Projects Releases Wiki Activity

working commit

Browse Source
This commit is contained in:
Олег Бородин
2026-02-04 22:55:48 +02:00
parent 3900d93559
commit afeab061e9
6 changed files with 117 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/handler/blob.go
+9 -8
View File
@@ -13,7 +13,7 @@ func (hand *Handler) DumpHeaders(message string, rctx *router.Context) {
hand.logg.Debugf("%s:\n%s\n", message, string(yamlData))
}
// 1 HEAD /v2/<name>/blobs/<digest> 200 404
// HEAD /v2/<name>/blobs/<digest> 200 404
func (hand *Handler) BlobExists(rctx *router.Context) {
name, _ := rctx.GetSubpath("name")
digest, _ := rctx.GetSubpath("digest")
@@ -34,7 +34,7 @@ func (hand *Handler) BlobExists(rctx *router.Context) {
rctx.SetStatus(code)
}
// 2 POST /v2/<name>/blobs/uploads/ 202 404
// POST /v2/<name>/blobs/uploads/ 202 404
func (hand *Handler) PostUpload(rctx *router.Context) {
name, _ := rctx.GetSubpath("name")
digest := rctx.GetQuery("digest")
@@ -63,10 +63,10 @@ func (hand *Handler) PostUpload(rctx *router.Context) {
// POST /v2/<name>/blobs/uploads/?digest=<digest> 201/202 404/400
// POST /v2/<name>/blobs/uploads/?mount=<digest>&from=<other_name> 201 404
// 3 PATCH /v2/<name>/blobs/uploads/<reference> 202 404/416
// PATCH /v2/<name>/blobs/uploads/<reference> 202 404/416
func (hand *Handler) PatchUpload(rctx *router.Context) {
hand.DumpHeaders("PatchUpload headers", rctx)
//hand.DumpHeaders("PatchUpload headers", rctx)
contentLength := rctx.GetHeader("Content-Length")
contentType := rctx.GetHeader("Content-Type")
@@ -90,15 +90,16 @@ func (hand *Handler) PatchUpload(rctx *router.Context) {
hand.logg.Errorf("PatchUpload error: %v", err)
}
rctx.SetHeader("Location", res.Location)
rctx.SetHeader("Range", res.Range)
rctx.SetStatus(code)
}
// 4 PUT /v2/<name>/blobs/uploads/<reference>?digest=<digest> 202 404/416
//
// PUT /v2/<name>/uploads/<reference>?digest=<digest> 202 404/416
// PUT /v2/<name>/blobs/uploads/<reference>?digest=<digest> 202 404/416
// PUT /v2/<name>/uploads/<reference>?digest=<digest> 202 404/416
func (hand *Handler) PutUpload(rctx *router.Context) {
hand.DumpHeaders("PutUpload headers", rctx)
//hand.DumpHeaders("PutUpload headers", rctx)
contentType := rctx.GetHeader("Content-Type")
contentLength := rctx.GetHeader("Content-Length")
app/handler/manifest.go
+12 -5
View File
@@ -87,14 +87,21 @@ func (hand *Handler) ManifestExists(rctx *router.Context) {
// PUT /v2/<name>/manifests/<reference> 201 404
func (hand *Handler) PutManifest(rctx *router.Context) {
hand.DumpHeaders("PutManifest headers", rctx)
contentType := rctx.GetHeader("Content-Type")
contentLength := rctx.GetHeader("Content-Length")
name, _ := rctx.GetSubpath("name")
reference, _ := rctx.GetSubpath("reference")
contentType := rctx.GetHeader("Content-Type")
params := &operator.PutManifestParams{
ContentType: contentType,
Name: name,
Reference: reference,
Reader: rctx.Request.Body,
ContentType: contentType,
ContentLength: contentLength,
Name: name,
Reference: reference,
Reader: rctx.Request.Body,
}
ctx := rctx.GetContext()
res, code, err := hand.oper.PutManifest(ctx, params)
app/maindb/manifest.go
+27 -42
View File
@@ -45,22 +45,27 @@ func (db *Database) InsertManifestWithLayers(ctx context.Context, manifest *desc
tx, err := db.db.BeginTx(ctx, nil)
// Manifest
request = `
INSERT INTO manifests(id, name, reference, contentType, payload, digest, created_at, updated_at)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8);`
INSERT INTO manifests(id, name, reference, contentType, payload, digest,
created_at, updated_at, created_by, updated_by)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10);`
_, err = tx.Exec(request, manifest.ID, manifest.Name, manifest.Reference,
manifest.ContentType, manifest.Payload, manifest.Digest,
manifest.CreatedAt, manifest.UpdatedAt)
manifest.CreatedAt, manifest.UpdatedAt,
manifest.CreatedBy, manifest.UpdatedBy)
if err != nil {
return err
}
// Layers
for _, layer := range layers {
request = `
INSERT INTO blobs(id, name, reference, mediaType, digest, size, created_at, updated_at)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8)`
INSERT INTO blobs(id, name, reference, mediaType, digest, size,
created_at, updated_at, created_by, updated_by)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)`
_, err = tx.Exec(request,
layer.ID, layer.Name, layer.Reference, layer.MediaType,
layer.Digest, layer.Size, layer.CreatedAt, layer.UpdatedAt)
layer.Digest, layer.Size,
layer.CreatedAt, layer.UpdatedAt,
layer.CreatedBy, layer.UpdatedBy)
if err != nil {
return err
}
@@ -74,7 +79,8 @@ func (db *Database) InsertManifestWithLayers(ctx context.Context, manifest *desc
return err
}
func (db *Database) UpdateManifestWithBlobs(ctx context.Context, manifest *descr.Manifest, newLayers []descr.Blob, oldLayers []descr.Blob) error {
func (db *Database) UpdateManifestWithBlobs(ctx context.Context, manifest *descr.Manifest,
newLayers []descr.Blob, oldLayers []descr.Blob) error {
var err error
var request string
@@ -83,20 +89,19 @@ func (db *Database) UpdateManifestWithBlobs(ctx context.Context, manifest *descr
// TODO: tx.Rollback()
// Manifest
request = `
UPDATE manifests
SET contentType = $1, payload = $2, digest = $3, updated_at = $4
WHERE name = $5 AND reference = $6`
request = `UPDATE manifests SET contentType = $1, payload = $2, digest = $3, updated_at = $4, updated_by = $5
WHERE name = $6 AND reference = $7`
_, err = tx.Exec(request, manifest.ContentType, manifest.Payload, manifest.Digest,
manifest.UpdatedAt, manifest.Name, manifest.Reference)
manifest.UpdatedAt, manifest.UpdatedBy,
manifest.Name, manifest.Reference)
if err != nil {
return err
}
// New layers
for _, layer := range newLayers {
request = `
INSERT INTO blobs(id, name, reference, mediaType, digest, size, created_at, updated_at)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8)`
request = `INSERT INTO blobs(id, name, reference, mediaType, digest, size,
created_at, updated_at, created_by, updated_by)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)`
_, err = tx.Exec(request, layer.ID, layer.Name, layer.Reference, layer.MediaType,
layer.Digest, layer.Size, layer.CreatedAt, layer.UpdatedAt)
if err != nil {
@@ -105,11 +110,7 @@ func (db *Database) UpdateManifestWithBlobs(ctx context.Context, manifest *descr
}
// Old layers
for _, layer := range oldLayers {
request = `
DELETE FROM blobs
WHERE name == $1
AND reference == $2
AND digest == $3`
request = `DELETE FROM blobs WHERE name == $1 AND reference == $2 AND digest == $3`
_, err = tx.Exec(request, layer.Name, layer.Reference, layer.Digest)
}
// Commit
@@ -124,11 +125,7 @@ func (db *Database) ManifestExistsByReference(ctx context.Context, name, referen
var err error
var count int
var exists bool
request := `
SELECT count(id) AS count FROM manifests
WHERE name = $1
AND reference = $2
LIMIT 1`
request := `SELECT count(id) AS count FROM manifests WHERE name = $1 AND reference = $2 LIMIT 1`
err = db.db.Get(&count, request, name, reference)
if err != nil {
return exists, err
@@ -143,9 +140,7 @@ func (db *Database) ManifestExistsByDigest(ctx context.Context, name, digest str
var err error
var count int
var exists bool
request := `
SELECT count(id) AS count FROM manifests
WHERE name = $1 AND digest = $2 LIMIT 1`
request := `SELECT count(id) AS count FROM manifests WHERE name = $1 AND digest = $2 LIMIT 1`
err = db.db.Get(&count, request, name, digest)
if err != nil {
return exists, err
@@ -161,10 +156,7 @@ func (db *Database) GetManifestByDigest(ctx context.Context, name, digest string
manifest := descr.Manifest{}
manifests := make([]descr.Manifest, 0)
exists := false
request := `
SELECT * FROM manifests
WHERE name = $1 AND digest = $2
LIMIT 1`
request := `SELECT * FROM manifests WHERE name = $1 AND digest = $2 LIMIT 1`
err = db.db.Select(&manifests, request, name, digest)
if err != nil {
return exists, manifest, err
@@ -181,10 +173,7 @@ func (db *Database) GetManifestByReference(ctx context.Context, name, reference
manifest := descr.Manifest{}
exists := false
manifests := make([]descr.Manifest, 0)
request := `
SELECT * FROM manifests
WHERE name = $1 AND reference = $2
LIMIT 1`
request := `SELECT * FROM manifests WHERE name = $1 AND reference = $2 LIMIT 1`
err = db.db.Select(&manifests, request, name, reference)
if err != nil {
return exists, manifest, err
@@ -200,9 +189,7 @@ func (db *Database) GetManifestByReference(ctx context.Context, name, reference
func (db *Database) ListManifestsByName(ctx context.Context, name string) ([]descr.Manifest, error) {
var err error
manifests := make([]descr.Manifest, 0)
request := `
SELECT id, name, reference, contentType, payload FROM manifests
WHERE name = $1`
request := `SELECT id, name, reference, contentType, payload FROM manifests WHERE name = $1`
err = db.db.Select(&manifests, request, name)
if err != nil {
return manifests, err
@@ -241,9 +228,7 @@ func (db *Database) DeleteManifest(ctx context.Context, name, reference string)
// TODO: tx.Rollback()
// Blobs
request = `
DELETE FROM blobs
WHERE name = $1 AND reference = $2`
request = `DELETE FROM blobs WHERE name = $1 AND reference = $2`
_, err = tx.Exec(request, name, reference)
if err != nil {
return err
app/operator/blob.go
+12 -1
View File
@@ -104,6 +104,7 @@ type PatchUploadParams struct {
}
type PatchUploadResult struct {
Location string
Range string
}
// The response for each successful chunk upload MUST be 202 Accepted, and MUST have the following headers:
@@ -125,6 +126,16 @@ func (oper *Operator) PatchUpload(ctx context.Context, params *PatchUploadParams
return res, http.StatusBadRequest, err
}
exists, uploadSize, err := oper.store.UploadExists(params.Name, params.Reference)
if err != nil {
return res, http.StatusInternalServerError, err
}
if exists {
res.Location = fmt.Sprintf("/v2/%s/uploads/%s", params.Name, params.Reference)
res.Range = fmt.Sprintf("0-%d", uploadSize-1)
return res, http.StatusNoContent, err
}
if params.ContentType != "application/octet-stream" {
err = fmt.Errorf("Wrong Conten-Type header: %s", params.ContentType)
return res, http.StatusBadRequest, err
@@ -150,7 +161,7 @@ func (oper *Operator) PatchUpload(ctx context.Context, params *PatchUploadParams
return res, http.StatusInternalServerError, err
}
res.Location = fmt.Sprintf("/v2/%s/uploads/%s", params.Name, params.Reference)
res.Range = fmt.Sprintf("0-%d", recsize-1)
return res, http.StatusAccepted, err
}
app/operator/manifest.go
+25 -8
View File
@@ -80,10 +80,11 @@ func (oper *Operator) ManifestExists(ctx context.Context, params *ManifestExists
}
type PutManifestParams struct {
ContentType string
Name string
Reference string
Reader io.Reader
ContentType string
ContentLength string
Name string
Reference string
Reader io.Reader
}
type PutManifestResult struct {
Location string
@@ -109,6 +110,17 @@ func (oper *Operator) PutManifest(ctx context.Context, params *PutManifestParams
err = fmt.Errorf("Unknown or empty Content-Type: %s", params.ContentType)
return res, http.StatusNotFound, err
}
if params.ContentLength == "" {
code := http.StatusLengthRequired
err = fmt.Errorf("Content-Length is empty")
return res, code, err
}
contentLength, err := strconv.ParseInt(params.ContentLength, 10, 64)
if err != nil {
err = fmt.Errorf("Cannot parse Content-Length value [%s]: %v", params.ContentLength, err)
code := http.StatusLengthRequired
return res, code, err
}
// Copy manifest data
buffer := bytes.NewBuffer(nil)
@@ -117,6 +129,13 @@ func (oper *Operator) PutManifest(ctx context.Context, params *PutManifestParams
return res, http.StatusInternalServerError, err
}
incomingManifestBytes := buffer.Bytes()
if int64(len(incomingManifestBytes)) != contentLength {
err = fmt.Errorf("Mismatch Content-Length and received manifest size: %d vs %d",
contentLength, len(incomingManifestBytes))
code := http.StatusInternalServerError
return res, code, err
}
oper.logg.Debugf("Manifest data: [%s]", string(incomingManifestBytes))
incomingManifest, err := auxoci.ParseOCIManifest(incomingManifestBytes)
@@ -124,10 +143,8 @@ func (oper *Operator) PutManifest(ctx context.Context, params *PutManifestParams
err = fmt.Errorf("Parsing OCI manifest error: %v", err)
return res, http.StatusInternalServerError, err
}
if incomingManifest.MediaType != params.ContentType {
err := fmt.Errorf("Mismatch Content-Type header and manifest MediaType: %s vs %s",
params.ContentType, incomingManifest.MediaType)
return res, http.StatusInternalServerError, err
if incomingManifest.MediaType == "" {
incomingManifest.MediaType = params.ContentType
}
manifestExists, _, err := oper.mdb.GetManifestByReference(ctx, params.Name, params.Reference)
app/storage/storage.go
+32 -1
View File
@@ -205,7 +205,6 @@ func (store *Storage) WriteUpload(uploadID string, source io.Reader) (int64, str
func (store *Storage) LinkUpload(reference, digest string) error {
var err error
uploadPath := store.makeUppath(reference)
blobPath := store.makeBlobpath(digest)
blobdir := store.makeBlobsubdir()
_, err = os.Stat(blobdir)
@@ -218,6 +217,20 @@ func (store *Storage) LinkUpload(reference, digest string) error {
if err != nil {
return err
}
blobPath := store.makeBlobpath(digest)
_, err = os.Stat(blobPath)
if err == nil {
err = os.Remove(blobPath)
if err != nil {
return err
}
}
if errors.Is(err, os.ErrNotExist) {
err = nil
}
if err != nil {
return err
}
err = os.Link(uploadPath, blobPath)
if err != nil {
@@ -240,6 +253,24 @@ func (store *Storage) RemoveUpload(digest string) error {
return err
}
func (st *Storage) UploadExists(name, reference string) (bool, int64, error) {
var err error
var fileSize int64
uploadPath := st.makeUppath(reference)
fileStat, err := os.Stat(uploadPath)
if errors.Is(err, os.ErrNotExist) {
return false, 0, nil
}
if err != nil {
return false, 0, err
}
fileSize = fileStat.Size()
return true, fileSize, err
}
func (store *Storage) WriteBlob(digest string, source io.Reader) (int64, string, error) {
var err error
var recsize int64