Feat/fetch p2p network info dfdaemon

apis/swagger.yml

@@ -1441,6 +1441,31 @@ definitions:
         items:
           $ref: "#/definitions/Node"
 
+  HeartBeatRequest:
+    type: "object"
+    description: ""
+    properties:
+      IP:
+        type: "string"
+        description: "IP address which peer client carries"
+        format: "string"
+      port:
+        type: "integer"
+        description: |
+          when registering, dfget will setup one uploader process.
+          This one acts as a server for peer pulling tasks.
+          This port is which this server listens on.
+        format: "int32"
+        minimum: 15000
+        maximum: 65000
+      cID:
+        type: "string"
+        description: |
+          CID means the client ID. It maps to the specific dfget process.
+          When user wishes to download an image/file, user would start a dfget process to do this.
+          This dfget is treated a client and carries a client ID.
+          Thus, multiple dfget processes on the same peer have different CIDs.
+
   ErrorResponse:
     type: "object"
     description: |

cmd/dfdaemon/app/root.go

@@ -18,6 +18,7 @@ package app
 
 import (
 	"encoding/json"
+	"github.com/dragonflyoss/Dragonfly/dfdaemon/localManager"
 	"os"
 	"os/exec"
 	"path/filepath"
@@ -68,6 +69,9 @@ var rootCmd = &cobra.Command{
 		cfgJSON, _ := json.Marshal(cfg)
 		logrus.Infof("using config: %s", cfgJSON)
 
+
+		_ = localManager.NewLocalManager(cfg.DFGetConfig())
+
 		s, err := dfdaemon.NewFromConfig(*cfg)
 		if err != nil {
 			return errors.Wrap(err, "create dfdaemon from config")

dfdaemon/config/config.go

@@ -113,6 +113,9 @@ type Properties struct {
 	LogConfig dflog.LogConfig `yaml:"logConfig" json:"logConfig"`
 	LocalIP   string          `yaml:"localIP" json:"localIP"`
 	PeerPort  int             `yaml:"peerPort" json:"peerPort"`
+
+	// Extreme is
+	Extreme	  *ExtremeConfig	`yaml:"extreme" json:"extreme"`
 }
 
 // Validate validates the config
@@ -145,13 +148,14 @@ func (p *Properties) DFGetConfig() DFGetConfig {
 	}
 
 	dfgetConfig := DFGetConfig{
-		DfgetFlags: dfgetFlags,
-		SuperNodes: p.SuperNodes,
-		RateLimit:  p.RateLimit.String(),
-		DFRepo:     p.DFRepo,
-		DFPath:     p.DFPath,
-		LocalIP:    p.LocalIP,
-		PeerPort:   p.PeerPort,
+		DfgetFlags:             dfgetFlags,
+		SuperNodes:             p.SuperNodes,
+		RateLimit:              p.RateLimit.String(),
+		DFRepo:                 p.DFRepo,
+		DFPath:                 p.DFPath,
+		LocalIP:                p.LocalIP,
+		PeerPort:               p.PeerPort,
+		SpecKeyOfExtremeTaskID: p.Extreme.SpecKeyOfTaskID,
 	}
 	if p.HijackHTTPS != nil {
 		dfgetConfig.HostsConfig = p.HijackHTTPS.Hosts
@@ -166,19 +170,25 @@ func (p *Properties) DFGetConfig() DFGetConfig {
 			})
 		}
 	}
+
+	if dfgetConfig.SpecKeyOfExtremeTaskID == "" {
+		dfgetConfig.SpecKeyOfExtremeTaskID = constant.DefaultSpecKeyOfExtremeTaskID
+	}
+
 	return dfgetConfig
 }
 
 // DFGetConfig configures how dfdaemon calls dfget.
 type DFGetConfig struct {
-	DfgetFlags  []string      `yaml:"dfget_flags"`
-	SuperNodes  []string      `yaml:"supernodes"`
-	RateLimit   string        `yaml:"ratelimit"`
-	DFRepo      string        `yaml:"localrepo"`
-	DFPath      string        `yaml:"dfpath"`
-	HostsConfig []*HijackHost `yaml:"hosts" json:"hosts"`
-	PeerPort    int           `yaml:"peerPort"`
-	LocalIP     string        `yaml:"localIP"`
+	DfgetFlags             []string      `yaml:"dfget_flags"`
+	SuperNodes             []string      `yaml:"supernodes"`
+	RateLimit              string        `yaml:"ratelimit"`
+	DFRepo                 string        `yaml:"localrepo"`
+	DFPath                 string        `yaml:"dfpath"`
+	HostsConfig            []*HijackHost `yaml:"hosts" json:"hosts"`
+	PeerPort               int           `yaml:"peerPort"`
+	LocalIP                string        `yaml:"localIP"`
+	SpecKeyOfExtremeTaskID string        `yaml:"specKeyOfTaskID"`
 }
 
 // RegistryMirror configures the mirror of the official docker registry
@@ -412,3 +422,11 @@ func NewProxy(regx string, useHTTPS bool, direct bool, redirect string) (*Proxy,
 func (r *Proxy) Match(url string) bool {
 	return r.Regx != nil && r.Regx.MatchString(url)
 }
+
+// ExtremeConfig represents the config of extreme mode
+type ExtremeConfig struct {
+	// SpecKeyOfExtremeTaskID defines the header key which represents the taskID
+	SpecKeyOfTaskID		string		`yaml:"specKeyOfTaskID" json:"specKeyOfTaskID"`
+
+	SpecKeyOfDirectRet string		`yaml:"specKeyOfDirectRet" json:"specKeyOfDirectRet"`
+}

dfdaemon/constant/constant.go

@@ -43,6 +43,10 @@ const (
 const (
 	// DefaultConfigPath is the default path of dfdaemon configuration file.
 	DefaultConfigPath = "/etc/dragonfly/dfdaemon.yml"
+
+	// DefaultSpecKeyOfExtremeTaskID is the default value of key which represents
+	// the taskID in extreme mode.
+	DefaultSpecKeyOfExtremeTaskID = "X-extreme-key"
 )
 
 const (

dfdaemon/downloader/p2p/dfclient.go

@@ -19,12 +19,9 @@ package p2p
 import (
 	"context"
 	"fmt"
-	"github.com/dragonflyoss/Dragonfly/pkg/httputils"
 	"io"
-	"net/http"
 	"os"
 	"path/filepath"
-	"strconv"
 	"strings"
 	"time"
 
@@ -43,11 +40,6 @@ type DFClient struct {
 	dfClient     core.DFGet
 }
 
-type DigestStruct struct {
-	Digest string
-	httputils.RangeStruct
-}
-
 func (c *DFClient) DownloadContext(ctx context.Context, url string, header map[string][]string, name string) (string, error) {
 	//	startTime := time.Now()
 	dstPath := filepath.Join(c.config.DFRepo, name)
@@ -107,27 +99,11 @@ func (c *DFClient) doDownload(ctx context.Context, url string, header map[string
 	runtimeConfig.URL = url
 	runtimeConfig.RV.TaskURL = url
 	runtimeConfig.RV.TaskFileName = getTaskFileName(destPath, c.dfGetConfig.Sign)
-	runtimeConfig.Header = flattenHeader(header)
+	runtimeConfig.Header = FlattenHeader(header)
 	runtimeConfig.Output = destPath
 	runtimeConfig.RV.RealTarget = destPath
 	runtimeConfig.RV.TargetDir = filepath.Dir(destPath)
 
-	hr := http.Header(header)
-	if digestHeaderStr := hr.Get(dfgetcfg.StrDigest); digestHeaderStr != "" {
-		ds, err := getDigest(digestHeaderStr)
-		if err != nil {
-			return nil, err
-		}
-
-		// todo: support the merge request
-		if len(ds) != 1 {
-			return nil, fmt.Errorf("no support to merge request")
-		}
-
-		runtimeConfig.RV.Digest = ds[0].Digest
-		runtimeConfig.RV.FileLength = (ds[0].EndIndex - ds[0].StartIndex) + 1
-	}
-
 	return c.dfClient.GetReader(ctx, &runtimeConfig)
 }
 
@@ -139,7 +115,7 @@ func getCid(localIP string, sign string) string {
 	return localIP + "-" + sign
 }
 
-func flattenHeader(header map[string][]string) []string {
+func FlattenHeader(header map[string][]string) []string {
 	var res []string
 	for key, value := range header {
 		// discard HTTP host header for backing to source successfully
@@ -156,61 +132,3 @@ func flattenHeader(header map[string][]string) []string {
 	}
 	return res
 }
-
-func getDigest(digestHeaderStr string) ([]*DigestStruct, error) {
-	var (
-		digest   string
-		rangeStr string
-	)
-
-	// digestHeaderStr looks like "sha256_1:0,1000;sha256_2:1001,2000"
-
-	result := []*DigestStruct{}
-
-	arr := strings.Split(digestHeaderStr, ";")
-	for _, elem := range arr {
-		kv := strings.Split(elem, ":")
-		if len(kv) > 3 || len(kv) < 2 {
-			return nil, fmt.Errorf("%s is not vaild for digestHeader", digestHeaderStr)
-		}
-
-		if len(kv) == 2 {
-			digest = fmt.Sprintf("sha256:%s", kv[0])
-			rangeStr = kv[1]
-		}
-
-		if len(kv) == 3 {
-			digest = fmt.Sprintf("%s:%s", kv[0], kv[1])
-			rangeStr = kv[2]
-		}
-
-		// todo: verify the sha256 string
-
-		rangeIndex := strings.Split(rangeStr, ",")
-		if len(rangeIndex) != 2 {
-			return nil, fmt.Errorf("%s is not vaild for digestHeader", digestHeaderStr)
-		}
-
-		startIndex, err := strconv.ParseInt(rangeIndex[0], 10, 64)
-		if err != nil {
-			return nil, fmt.Errorf("%s is not vaild for digestHeader", digestHeaderStr)
-		}
-
-		endIndex, err := strconv.ParseInt(rangeIndex[1], 10, 64)
-		if err != nil {
-			return nil, fmt.Errorf("%s is not vaild for digestHeader", digestHeaderStr)
-		}
-
-		ds := &DigestStruct{
-			Digest: digest,
-			RangeStruct: httputils.RangeStruct{
-				StartIndex: startIndex,
-				EndIndex:   endIndex,
-			},
-		}
-
-		result = append(result, ds)
-	}
-
-	return result, nil
-}

dfdaemon/downloader/p2p/dfclient_test.go

@@ -63,7 +63,7 @@ func (tc *testcase) WithValue(key string, value ...string) *testcase {
 func (tc *testcase) Test(t *testing.T) {
 	a := assert.New(t)
 
-	actualResult := flattenHeader(tc.header)
+	actualResult := FlattenHeader(tc.header)
 
 	for _, r := range actualResult {
 		if _, ok := tc.result[r]; !ok {

dfdaemon/localManager/cache.go

@@ -0,0 +1,12 @@
+package localManager
+
+import "github.com/dragonflyoss/Dragonfly/pkg/syncmap"
+
+type cache struct {
+
+}
+
+type cacheManager struct {
+	// key is taskID, value is the
+	taskContainer	*syncmap.SyncMap
+}