mstore/app/operator/blob.go

package operator

import (
	"context"
	"fmt"
	"io"
	"net/http"
	"strconv"

	"mstore/pkg/auxuuid"
)

type BlobExistsParams struct {
	Name   string
	Digest string
}
type BlobExistsResult struct {
	DockerContentDigest string
	ContentLength       string
	ContentType         string
	//Exists              bool
}

func (oper *Operator) BlobExists(ctx context.Context, params *BlobExistsParams) (*BlobExistsResult, int, error) {
	var err error
	res := &BlobExistsResult{}

	if params.Digest == "" {
		err = fmt.Errorf("Empty reference")
		return res, http.StatusBadRequest, err
	}
	if params.Name == "" {
		err = fmt.Errorf("Empty name")
		return res, http.StatusBadRequest, err
	}
	// Check blob descriptor
	descrExists, blobDescr, err := oper.mdb.GetBlobByNameDigest(ctx, params.Digest, params.Digest)
	if err != nil {
		return res, http.StatusInternalServerError, err
	}
	if !descrExists {
		return res, http.StatusNotFound, err
	}
	// Check blob file
	blobExists, _, err := oper.store.BlobExists(params.Digest)
	if err != nil {
		return res, http.StatusInternalServerError, err
	}
	if !blobExists {
		return res, http.StatusNotFound, err
	}

	res.ContentLength = strconv.FormatInt(blobDescr.Size, 10)
	res.DockerContentDigest = blobDescr.Digest
	res.ContentType = blobDescr.MediaType

	return res, http.StatusOK, err
}

type PostUploadParams struct {
	Name   string
	Digest string
	Mount  string
	From   string
}
type PostUploadResult struct {
	DockerUploadUUID string
	Location         string
	ContentLength    string
}

func (oper *Operator) PostUpload(ctx context.Context, params *PostUploadParams) (*PostUploadResult, int, error) {
	var err error
	res := &PostUploadResult{}

	if params.Digest == "" {
		uuid := auxuuid.NewUUID()
		location := fmt.Sprintf("/v2/%s/blobs/uploads/%s", params.Name, uuid)
		res.DockerUploadUUID = uuid
		res.Location = location
		res.ContentLength = strconv.FormatInt(0, 10)
		return res, http.StatusAccepted, err
	} else {
		err = fmt.Errorf("PostUpload: Not empty digest header")
		return res, http.StatusInternalServerError, err
	}
	return res, http.StatusOK, err
}

type PatchUploadParams struct {
	ContentType   string
	ContentLength string
	ContentRange  string
	Name          string
	Reference     string
	Reader        io.Reader
}
type PatchUploadResult struct {
	Location string
	Range    string
}

// TODO: partial uploading by range?
func (oper *Operator) PatchUpload(ctx context.Context, params *PatchUploadParams) (*PatchUploadResult, int, error) {
	var err error
	res := &PatchUploadResult{}

	if params.Reference == "" {
		err = fmt.Errorf("Empty reference")
		return res, http.StatusBadRequest, err
	}
	if params.Name == "" {
		err = fmt.Errorf("Empty name")
		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
	}
	var contentLength int64

	// Unfortunately, podman and used github.com/containers/image don't sent
	// Content-length header for docker transport

	if params.ContentLength != "" {
		contentLength, err = strconv.ParseInt(params.ContentLength, 10, 64)
		if err != nil {
			err = fmt.Errorf("Wrong Content-length header")
			return res, http.StatusBadRequest, err
		}
	}

	recsize, _, err := oper.store.WriteUpload(params.Reference, params.Reader)
	if err != nil {
		return res, http.StatusInternalServerError, err
	}
	if contentLength != 0 && recsize != contentLength {
		oper.store.RemoveUpload(params.Reference)
		err = fmt.Errorf("Mismatch upload recorded size and content length")
		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
}

type PutUploadParams struct {
	ContentType   string
	ContentLength string
	ContentRange  string
	Name          string
	Reference     string
	Digest        string
	Reader        io.Reader
}
type PutUploadResult struct {
	Location string
}

func (oper *Operator) PutUpload(ctx context.Context, params *PutUploadParams) (*PutUploadResult, int, error) {
	var err error
	res := &PutUploadResult{}

	if params.Reference == "" {
		err = fmt.Errorf("Empty reference")
		return res, http.StatusBadRequest, err
	}
	if params.Name == "" {
		err = fmt.Errorf("Empty name")
		return res, http.StatusBadRequest, err
	}
	if params.Digest == "" {
		err = fmt.Errorf("Empty digest")
		return res, http.StatusBadRequest, err
	}
	if params.ContentType != "application/octet-stream" {
		err = fmt.Errorf("Wrong conten type: %s", params.ContentType)
		return res, http.StatusBadRequest, err
	}

	var contentLength int64
	if params.ContentLength != "" {
		contentLength, err = strconv.ParseInt(params.ContentLength, 10, 64)
		if err != nil {
			err = fmt.Errorf("Cannot convert Content-Length=%s to integer: %v", params.ContentLength, err)
			return res, http.StatusBadRequest, err
		}

	}
	if contentLength != 0 {
		// TODO
		err = fmt.Errorf("Unexpected Content-Length header: %s", params.ContentLength)
		return res, http.StatusInternalServerError, err
	}

	err = oper.store.LinkUpload(params.Reference, params.Digest)
	if err != nil {
		err = fmt.Errorf("Failed to link upload %s, err: %v", params.Reference, err)
		return res, http.StatusInternalServerError, err
	}

	res.Location = fmt.Sprintf("/v2/%s/blobs/%s", params.Name, params.Digest)
	return res, http.StatusCreated, err
}

type GetBlobParams struct {
	Name   string
	Digest string
}
type GetBlobResult struct {
	ContentLength       string
	ContentType         string
	DockerContentDigest string
	ReadCloser          io.ReadCloser
}

func (oper *Operator) GetBlob(ctx context.Context, params *GetBlobParams) (*GetBlobResult, int, error) {
	var err error
	res := &GetBlobResult{}

	if params.Name == "" {
		err = fmt.Errorf("Empty name")
		return res, http.StatusBadRequest, err
	}
	if params.Digest == "" {
		err = fmt.Errorf("Empty digest")
		return res, http.StatusBadRequest, err
	}

	blobExists, blobSize, err := oper.store.BlobExists(params.Digest)
	if err != nil {
		return res, http.StatusInternalServerError, err
	}
	descrExist, blobDescr, err := oper.mdb.GetBlobByNameDigest(ctx, params.Name, params.Digest)
	if err != nil {
		return res, http.StatusInternalServerError, err
	}
	if !blobExists || !descrExist {
		return res, http.StatusNotFound, err
	}

	_, readCloser, err := oper.store.BlobReader(params.Digest)
	if err != nil {
		return res, http.StatusInternalServerError, err
	}

	res.ContentType = blobDescr.MediaType
	res.ContentLength = strconv.FormatInt(blobSize, 10)
	res.DockerContentDigest = params.Digest
	res.ReadCloser = readCloser
	return res, http.StatusOK, err
}

type DeleteBlobParams struct {
	Name   string
	Digest string
}
type DeleteBlobResult struct{}

// Removing an individual layer is very probably to compromise data integrity.
// - If the data layers are to be reused, this is like shooting yourself in the foot.
// - It also prevents the manifest from being issued, as one
// of the layers is missing.

func (oper *Operator) DeleteBlob(ctx context.Context, params *DeleteBlobParams) (*DeleteBlobResult, int, error) {
	var err error
	res := &DeleteBlobResult{}

	if params.Digest == "" {
		err = fmt.Errorf("Empty digest")
		return res, http.StatusBadRequest, err
	}
	if params.Name == "" {
		err = fmt.Errorf("Empty name")
		return res, http.StatusBadRequest, err
	}
	// Check namespace record
	descrExists, _, err := oper.mdb.GetBlobByNameDigest(ctx, params.Name, params.Digest)
	if err != nil {
		return res, http.StatusInternalServerError, err
	}

	if !descrExists {
		return res, http.StatusNotFound, err
	}
	// Deleting blob record
	oper.logg.Warningf("Deleting blob record %s:%s", params.Name, params.Digest)
	err = oper.mdb.DeleteBlobByNameDigest(ctx, params.Name, params.Digest)
	if err != nil {
		return res, http.StatusInternalServerError, err
	}
	// Removing the blob binary if usage == 0
	blobUsage, err := oper.mdb.GetBlobUsage(ctx, params.Digest)
	if err != nil {
		return res, http.StatusInternalServerError, err
	}
	if blobUsage == 0 {
		oper.logg.Warningf("Deleting useless blob binary %s", params.Digest)
		oper.store.DeleteBlob(params.Digest)
	}
	return res, http.StatusOK, err
}