preheat support manifests list

dragonflyoss · Aug 25, 2020 · 25c8fa3 · 25c8fa3
1 parent ce14eb4
commit 25c8fa3
Show file tree

Hide file tree

Showing 8 changed files with 33 additions and 12 deletions.
diff --git a/apis/types/preheat_info.go b/apis/types/preheat_info.go
diff --git a/supernode/daemon/mgr/preheat/base_worker.go b/supernode/daemon/mgr/preheat/base_worker.go
@@ -24,7 +24,7 @@ import (
 	"github.com/dragonflyoss/Dragonfly/supernode/daemon/mgr"
 )
 
-const TIMEOUT = 30 * 60;
+const TIMEOUT = 30 * 60
 
 var _ IWorker = &BaseWorker{}
 

diff --git a/supernode/daemon/mgr/preheat/file_preaheater.go b/supernode/daemon/mgr/preheat/file_preaheater.go
@@ -17,6 +17,7 @@ package preheat
 
 import (
 	"errors"
+	"fmt"
 	"github.com/sirupsen/logrus"
 	"time"
 
@@ -87,7 +88,7 @@ func (w *FileWorker) query() chan error {
 			status := w.progress.cmd.ProcessState
 			if status != nil && status.Exited() {
 				if !status.Success() {
-					errMsg := "dfget failed:" + status.String()
+					errMsg := fmt.Sprintf("dfget failed: %s err: %s",  status.String(), w.progress.errmsg.String())
 					w.failed(errMsg)
 					w.Preheater.Cancel(w.Task.ID)
 					result <- errors.New(errMsg)

diff --git a/supernode/daemon/mgr/preheat/image_preaheater.go b/supernode/daemon/mgr/preheat/image_preaheater.go
@@ -95,6 +95,7 @@ func (w *ImageWorker) query() chan error {
 					errMsg := childTask.URL + " " + childTask.ErrorMsg
 					w.Preheater.Cancel(w.Task.ID)
 					result <- errors.New(errMsg)
+					logrus.Errorf("PreheatImage Task [%s] prehead failed for ", w.Task.ID, errMsg)
 					return
 				}
 			}
@@ -230,7 +231,18 @@ func (w *ImageWorker) parseLayers(body []byte, header map[string]string) (layers
 	if schemaVersion == "1" {
 		layerDigest = w.parseLayerDigest(meta, "fsLayers", "blobSum")
 	} else {
-		layerDigest = w.parseLayerDigest(meta, "layers", "digest")
+		mediaType := fmt.Sprintf("%s", meta["mediaType"])
+		switch mediaType {
+		case "application/vnd.docker.distribution.manifest.list.v2+json", "application/vnd.oci.image.index.v1+json":
+			manifestDigest := w.parseLayerDigest(meta, "manifests", "digest")
+			for _, digest := range manifestDigest {
+				list, _ := w.getLayers(w.manifestUrl(digest), header, false)
+				layers = append(layers, list...)
+			}
+			return
+		default:
+			layerDigest = w.parseLayerDigest(meta, "layers", "digest")
+		}
 	}
 
 	for _, digest := range layerDigest {
@@ -248,6 +260,10 @@ func (w *ImageWorker) layerUrl(digest string) string {
 	return fmt.Sprintf("%s://%s/v2/%s/blobs/%s", w.protocol, w.domain, w.name, digest)
 }
 
+func (w *ImageWorker) manifestUrl(digest string) string {
+	return fmt.Sprintf("%s://%s/v2/%s/manifests/%s", w.protocol, w.domain, w.name, digest)
+}
+
 func (w *ImageWorker) parseLayerDigest(meta map[string]interface{}, layerKey string, digestKey string) (layers []string) {
 	data := meta[layerKey]
 	if data == nil {

diff --git a/supernode/daemon/mgr/preheat/manager.go b/supernode/daemon/mgr/preheat/manager.go
@@ -18,6 +18,7 @@ package preheat
 
 import (
 	"context"
+	"github.com/sirupsen/logrus"
 	"net/http"
 
 	"github.com/dragonflyoss/Dragonfly/apis/types"
@@ -44,6 +45,8 @@ func (m *Manager) Create(ctx context.Context, task *types.PreheatCreateRequest)
 	preheatTask.Filter = task.Filter
 	preheatTask.Identifier = task.Identifier
 	preheatTask.Headers = task.Headers
+	logrus.Debugf("create preheat: Type[%s] URL[%s] Filter[%s] Identifier[%s] Headers[%v]",
+		preheatTask.Type, preheatTask.URL, preheatTask.Filter, preheatTask.Identifier, preheatTask.Headers)
 	return m.service.Create(preheatTask)
 }
 

diff --git a/supernode/daemon/mgr/preheat/preheat_progress.go b/supernode/daemon/mgr/preheat/preheat_progress.go
@@ -16,18 +16,23 @@
 package preheat
 
 import (
+	"bytes"
 	"os/exec"
 )
 
 type PreheatProgress struct {
 	output string
 	cmd *exec.Cmd
+	errmsg *bytes.Buffer
 }
 
 func NewPreheatProgress(output string, cmd *exec.Cmd) *PreheatProgress {
 	p := &PreheatProgress{
 		output: output,
 		cmd: cmd,
+		errmsg: bytes.NewBuffer(make([]byte, 0, 128)),
 	}
+	cmd.Stderr = p.errmsg
+	cmd.Stdout = p.errmsg
 	return p
 }
diff --git a/supernode/daemon/mgr/preheat/preheat_task_repository.go b/supernode/daemon/mgr/preheat/preheat_task_repository.go
@@ -23,20 +23,18 @@ import (
 )
 
 const(
+	// preheat image cache one week
 	EXPIRED_TIME = 7 * 24 * 3600 * 1000;
 )
 
 type PreheatTaskRepository struct {
 	preheatTasks *sync.Map
-	loaded *sync.Once
 }
 
 func NewPreheatTaskRepository() *PreheatTaskRepository {
 	r := &PreheatTaskRepository{
 		preheatTasks: new(sync.Map),
-		loaded: new(sync.Once),
 	}
-	r.load()
 	return r
 }
 
@@ -113,9 +111,3 @@ func (r *PreheatTaskRepository) expired(timestamp int64) bool {
 	return time.Now().UnixNano()/int64(time.Millisecond) > timestamp+EXPIRED_TIME
 }
 
-func(r *PreheatTaskRepository) load() {
-	r.loaded.Do(func(){
-
-	})
-}
-
diff --git a/supernode/server/preheat_bridge.go b/supernode/server/preheat_bridge.go
@@ -64,6 +64,7 @@ func (s *Server) getPreheatTask(ctx context.Context, rw http.ResponseWriter, req
 		FinishTime: strfmt.DateTime(time.Unix(task.FinishTime/1000, task.FinishTime%1000*int64(time.Millisecond)).UTC()),
 		StartTime:  strfmt.DateTime(time.Unix(task.StartTime/1000, task.StartTime%1000*int64(time.Millisecond)).UTC()),
 		Status:     task.Status,
+		ErrorMsg:   task.ErrorMsg,
 	}
 	return EncodeResponse(rw, http.StatusOK, resp)
 }