test: dump goroutine in e2e

client/daemon/peer/peertask_conductor.go

@@ -702,7 +702,7 @@ loop:
 		}
 
 		// update content length
-		if piecePacket.ContentLength > 0 {
+		if piecePacket.ContentLength > -1 {
 			pt.SetContentLength(piecePacket.ContentLength)
 			_ = pt.UpdateStorage(false)
 			pt.Debugf("update content length: %d", pt.GetContentLength())
@@ -731,7 +731,7 @@ loop:
 
 func (pt *peerTaskConductor) init(piecePacket *base.PiecePacket, pieceBufferSize uint32) (chan *DownloadPieceRequest, bool) {
 	pt.contentLength.Store(piecePacket.ContentLength)
-	if piecePacket.ContentLength > 0 {
+	if piecePacket.ContentLength > -1 {
 		pt.span.SetAttributes(config.AttributeTaskContentLength.Int64(piecePacket.ContentLength))
 	}
 	if err := pt.InitStorage(); err != nil {

client/daemon/peer/peertask_manager.go

@@ -169,6 +169,8 @@ func (ptm *peerTaskManager) getOrCreatePeerTaskConductor(ctx context.Context, ta
 	// double check
 	if p, ok := ptm.findPeerTaskConductor(taskID); ok {
 		ptm.conductorLock.Unlock()
+		logger.Debugf("same peer task found: %s/%s, cancel created peer task %s/%s",
+			p.taskID, p.peerID, ptc.taskID, ptc.peerID)
 		// cancel duplicate peer task
 		ptc.cancel(base.Code_ClientContextCanceled, reasonContextCanceled)
 		return p, nil
@@ -231,7 +233,7 @@ func (ptm *peerTaskManager) Stop(ctx context.Context) error {
 }
 
 func (ptm *peerTaskManager) PeerTaskDone(taskID string) {
-	logger.Debugf("delete task %s in running tasks", taskID)
+	logger.Debugf("delete done task %s in running tasks", taskID)
 	ptm.runningPeerTasks.Delete(taskID)
 }
 

client/daemon/proxy/proxy.go

@@ -343,7 +343,12 @@ func (proxy *Proxy) handleHTTP(span trace.Span, w http.ResponseWriter, req *http
 	w.WriteHeader(resp.StatusCode)
 	span.SetAttributes(semconv.HTTPStatusCodeKey.Int(resp.StatusCode))
 	if n, err := io.Copy(w, resp.Body); err != nil && err != io.EOF {
-		logger.Errorf("failed to write http body: %v", err)
+		if peerID := resp.Header.Get(config.HeaderDragonflyPeer); peerID != "" {
+			logger.Errorf("failed to write http body: %v, peer: %s, task: %s",
+				err, peerID, resp.Header.Get(config.HeaderDragonflyTask))
+		} else {
+			logger.Errorf("failed to write http body: %v", err)
+		}
 		span.RecordError(err)
 	} else {
 		span.SetAttributes(semconv.HTTPResponseContentLengthKey.Int64(n))

test/e2e/constants.go

@@ -40,6 +40,7 @@ type server struct {
 	namespace  string
 	logDirName string
 	replicas   int
+	pprofPort  int
 }
 
 var servers = map[string]server{
@@ -66,11 +67,13 @@ var servers = map[string]server{
 		namespace:  dragonflyNamespace,
 		logDirName: "daemon",
 		replicas:   1,
+		pprofPort:  9999,
 	},
 	proxyServerName: {
 		name:       proxyServerName,
 		namespace:  dragonflyE2ENamespace,
 		logDirName: "daemon",
 		replicas:   3,
+		pprofPort:  9999,
 	},
 }

test/e2e/dfget_test.go

@@ -59,6 +59,7 @@ func singleDfgetTest(name, ns, label, podNamePrefix, container string) {
 		}
 
 		for url, path := range urls {
+			fmt.Printf("--------------------------------------------------------------------------------\n\n")
 			fmt.Println("download url: " + url)
 			// get original file digest
 			out, err = e2eutil.DockerCommand("sha256sum", path).CombinedOutput()

test/e2e/e2e_test.go

@@ -59,7 +59,7 @@ var _ = AfterSuite(func() {
 			out, err = pod.Command("sh", "-c", fmt.Sprintf(`
               set -x
               cp -r /var/log/dragonfly/%s /tmp/artifact/%s-%d
-							find /tmp/artifact -type d -exec chmod 777 {} \;
+              find /tmp/artifact -type d -exec chmod 777 {} \;
               `, server.logDirName, server.name, i)).CombinedOutput()
 			if err != nil {
 				fmt.Printf("copy log output: %s, error: %s\n", string(out), err)
@@ -75,6 +75,13 @@ var _ = AfterSuite(func() {
 				fmt.Printf("upload pod %s artifact prev stdout file error: %v\n", podName, err)
 			}
 
+			if server.pprofPort > 0 {
+				if out, err := e2eutil.UploadArtifactPProf(server.namespace, podName,
+					fmt.Sprintf("%s-%d", server.name, i), server.name, server.pprofPort); err != nil {
+					fmt.Printf("upload pod %s artifact pprof error: %v, output: %s\n", podName, err, out)
+				}
+			}
+
 		}
 	}
 })

test/e2e/e2eutil/artifact.go

@@ -66,3 +66,17 @@ func UploadArtifactPrevStdout(namespace, podName, logDirName, logPrefix string)
 
 	return nil
 }
+
+func UploadArtifactPProf(namespace, podName, logDirName, logPrefix string, pprofPort int) (string, error) {
+	logDirname := fmt.Sprintf("/tmp/artifact/%s/", logDirName)
+	out, err := KubeCtlCommand("-n", namespace, "exec", podName, "--", "sh", "-c", fmt.Sprintf(`
+      set -x
+      port=%d
+      dir=%s
+      prefix=%s
+      ip=$(hostname -i)
+      wget $ip:$port/debug/pprof/"goroutine?debug=1" -O $dir/$prefix-pprof-goroutine-1.log
+      wget $ip:$port/debug/pprof/"goroutine?debug=2" -O $dir/$prefix-pprof-goroutine-2.log
+      `, pprofPort, logDirname, logPrefix)).CombinedOutput()
+	return string(out), err
+}

test/testdata/charts/config.yaml

@@ -33,6 +33,8 @@ dfdaemon:
     hostPath:
       path: /tmp/artifact
   config:
+    verbose: true
+    pprofPort: 9999
     scheduler:
       disableAutoBackSource: true
     proxy:

test/testdata/k8s/proxy.yaml

@@ -7,9 +7,9 @@ data:
   dfget.yaml: |-
     aliveTime: 0s
     gcInterval: 1m0s
-    keepStorage: false
+    keepStorage: true
     verbose: true
-    pprof-port: 0
+    pprof-port: 9999
     scheduler:
       manager:
         enable: true