From f8efb7953eac604d306c3d79e60f1b2ad2916aa9 Mon Sep 17 00:00:00 2001 From: Michael Yang Date: Wed, 8 May 2024 17:34:54 -0700 Subject: [PATCH] update push to use model.Name --- server/images.go | 31 +++++++++++++++---------------- server/routes.go | 26 ++++++++------------------ server/upload.go | 39 +++++++++++++++++++++++---------------- 3 files changed, 46 insertions(+), 50 deletions(-) diff --git a/server/images.go b/server/images.go index 8848011d94..5330a924bc 100644 --- a/server/images.go +++ b/server/images.go @@ -602,43 +602,42 @@ func CopyModel(src, dst model.Name) error { return err } -func PushModel(ctx context.Context, name string, regOpts *registryOptions, fn func(api.ProgressResponse)) error { - mp := ParseModelPath(name) +func PushModel(ctx context.Context, name model.Name, opts registryOptions, fn func(api.ProgressResponse)) error { fn(api.ProgressResponse{Status: "retrieving manifest"}) - if mp.ProtocolScheme == "http" && !regOpts.Insecure { - return fmt.Errorf("insecure protocol http") + m, err := ParseNamedManifest(name) + if err != nil { + return err + } + + scheme := "https" + if opts.Insecure { + scheme = "http" } - manifest, _, err := GetManifest(mp) + baseURL, err := url.Parse(fmt.Sprintf("%s://%s", scheme, path.Join(name.Host, "v2", name.Namespace, name.Model))) if err != nil { - fn(api.ProgressResponse{Status: "couldn't retrieve manifest"}) return err } - var layers []*Layer - layers = append(layers, manifest.Layers...) - layers = append(layers, manifest.Config) - - for _, layer := range layers { - if err := uploadBlob(ctx, mp, layer, regOpts, fn); err != nil { + for _, layer := range append(m.Layers, m.Config) { + if err := uploadBlob(ctx, uploadOptions{name: name, baseURL: baseURL, layer: layer, regOpts: &opts, fn: fn}); err != nil { slog.Info(fmt.Sprintf("error uploading blob: %v", err)) return err } } fn(api.ProgressResponse{Status: "pushing manifest"}) - requestURL := mp.BaseURL() - requestURL = requestURL.JoinPath("v2", mp.GetNamespaceRepository(), "manifests", mp.Tag) + requestURL := baseURL.JoinPath("v2", name.Namespace, name.Model, "manifests", name.Tag) - manifestJSON, err := json.Marshal(manifest) + manifestJSON, err := json.Marshal(m) if err != nil { return err } headers := make(http.Header) headers.Set("Content-Type", "application/vnd.docker.distribution.manifest.v2+json") - resp, err := makeRequestWithRetry(ctx, http.MethodPut, requestURL, headers, bytes.NewReader(manifestJSON), regOpts) + resp, err := makeRequestWithRetry(ctx, http.MethodPut, requestURL, headers, bytes.NewReader(manifestJSON), &opts) if err != nil { return err } diff --git a/server/routes.go b/server/routes.go index 13d40a79d6..da95b46f8d 100644 --- a/server/routes.go +++ b/server/routes.go @@ -447,24 +447,18 @@ func (s *Server) PullModelHandler(c *gin.Context) { } func (s *Server) PushModelHandler(c *gin.Context) { - var req api.PushRequest - err := c.ShouldBindJSON(&req) - switch { - case errors.Is(err, io.EOF): + var r api.PushRequest + if err := c.ShouldBindJSON(&r); errors.Is(err, io.EOF) { c.AbortWithStatusJSON(http.StatusBadRequest, gin.H{"error": "missing request body"}) return - case err != nil: + } else if err != nil { c.AbortWithStatusJSON(http.StatusBadRequest, gin.H{"error": err.Error()}) return } - var model string - if req.Model != "" { - model = req.Model - } else if req.Name != "" { - model = req.Name - } else { - c.AbortWithStatusJSON(http.StatusBadRequest, gin.H{"error": "model is required"}) + n := model.ParseName(cmp.Or(r.Model, r.Name)) + if !n.IsValid() { + c.AbortWithStatusJSON(http.StatusBadRequest, gin.H{"error": fmt.Sprintf("name %q is invalid", cmp.Or(r.Model, r.Name))}) return } @@ -475,19 +469,15 @@ func (s *Server) PushModelHandler(c *gin.Context) { ch <- r } - regOpts := ®istryOptions{ - Insecure: req.Insecure, - } - ctx, cancel := context.WithCancel(c.Request.Context()) defer cancel() - if err := PushModel(ctx, model, regOpts, fn); err != nil { + if err := PushModel(ctx, n, registryOptions{Insecure: r.Insecure}, fn); err != nil { ch <- gin.H{"error": err.Error()} } }() - if req.Stream != nil && !*req.Stream { + if r.Stream != nil && !*r.Stream { waitForStream(c, ch) return } diff --git a/server/upload.go b/server/upload.go index 9b52238a40..f103f6306f 100644 --- a/server/upload.go +++ b/server/upload.go @@ -18,6 +18,7 @@ import ( "github.com/ollama/ollama/api" "github.com/ollama/ollama/format" + "github.com/ollama/ollama/types/model" "golang.org/x/sync/errgroup" ) @@ -360,40 +361,46 @@ func (p *progressWriter) Rollback() { p.written = 0 } -func uploadBlob(ctx context.Context, mp ModelPath, layer *Layer, opts *registryOptions, fn func(api.ProgressResponse)) error { - requestURL := mp.BaseURL() - requestURL = requestURL.JoinPath("v2", mp.GetNamespaceRepository(), "blobs", layer.Digest) +type uploadOptions struct { + name model.Name + baseURL *url.URL + layer *Layer + regOpts *registryOptions + fn func(api.ProgressResponse) +} + +func uploadBlob(ctx context.Context, opts uploadOptions) error { + requestURL := opts.baseURL.JoinPath("v2", opts.name.Namespace, opts.name.Model, "blobs", opts.layer.Digest) - resp, err := makeRequestWithRetry(ctx, http.MethodHead, requestURL, nil, nil, opts) + resp, err := makeRequestWithRetry(ctx, http.MethodHead, requestURL, nil, nil, opts.regOpts) switch { case errors.Is(err, os.ErrNotExist): case err != nil: return err default: defer resp.Body.Close() - fn(api.ProgressResponse{ - Status: fmt.Sprintf("pushing %s", layer.Digest[7:19]), - Digest: layer.Digest, - Total: layer.Size, - Completed: layer.Size, + opts.fn(api.ProgressResponse{ + Status: fmt.Sprintf("pushing %s", opts.layer.Digest[7:19]), + Digest: opts.layer.Digest, + Total: opts.layer.Size, + Completed: opts.layer.Size, }) return nil } - data, ok := blobUploadManager.LoadOrStore(layer.Digest, &blobUpload{Layer: layer}) + data, ok := blobUploadManager.LoadOrStore(opts.layer.Digest, &blobUpload{Layer: opts.layer}) upload := data.(*blobUpload) if !ok { - requestURL := mp.BaseURL() - requestURL = requestURL.JoinPath("v2", mp.GetNamespaceRepository(), "blobs/uploads/") - if err := upload.Prepare(ctx, requestURL, opts); err != nil { - blobUploadManager.Delete(layer.Digest) + requestURL := opts.baseURL.JoinPath("v2", opts.name.Namespace, opts.name.Model, "blobs/uploads/") + if err := upload.Prepare(ctx, requestURL, opts.regOpts); err != nil { + blobUploadManager.Delete(opts.layer.Digest) return err } // nolint: contextcheck - go upload.Run(context.Background(), opts) + go upload.Run(context.Background(), opts.regOpts) } - return upload.Wait(ctx, fn) + return upload.Wait(ctx, opts.fn) }