Merge pull request #1717 from 0chain/feat/timings-logs

Add timings logs for download blocks

core/sys/fs.go

@@ -41,6 +41,8 @@ type FS interface {
 	// RemoveProgress remove progress
 	RemoveProgress(progressID string) error
 
+	StoreLogs(key string, data string) error
+
 	// Create Directory
 	CreateDirectory(dirID string) error
 

core/sys/fs_disk.go

@@ -72,6 +72,10 @@ func (dfs *DiskFS) RemoveProgress(progressID string) error {
 	return dfs.Remove(progressID)
 }
 
+func (dfs *DiskFS) StoreLogs(key string, data string) error {
+	return nil
+}
+
 func (dfs *DiskFS) CreateDirectory(_ string) error {
 	return nil
 }

core/sys/fs_mem.go

@@ -4,6 +4,7 @@
 package sys
 
 import (
+	"encoding/json"
 	"errors"
 	"io/fs"
 	"os"
@@ -152,6 +153,24 @@ func (mfs *MemFS) RemoveProgress(progressID string) error {
 	return nil
 }
 
+func (mfs *MemFS) StoreLogs(key string, data string) error {
+	//get existing logs
+	var logs []string
+	val := js.Global().Get("localStorage").Call("getItem", key)
+	if val.Truthy() {
+		json.Unmarshal([]byte(val.String()), &logs)
+	}
+	//append new logs
+	logs = append(logs, data)
+	//store logs
+	encodedData, err := json.Marshal(logs)
+	if err != nil {
+		return err
+	}
+	js.Global().Get("localStorage").Call("setItem", key, string(encodedData))
+	return nil
+}
+
 func (mfs *MemFS) CreateDirectory(dirID string) error {
 	if !js.Global().Get("showDirectoryPicker").Truthy() || !js.Global().Get("WritableStream").Truthy() {
 		return errors.New("dir_picker: not supported")

zboxcore/sdk/allocation.go

@@ -50,6 +50,8 @@ var (
 	MultiOpBatchSize = 50
 	RepairBatchSize  = 50
 	Workdir          string
+	logChanMap       = make(map[string]chan logEntry)
+	logMapMutex      = &sync.Mutex{}
 )
 
 const (
@@ -437,6 +439,9 @@ func (a *Allocation) InitAllocation() {
 			}
 		}
 	}
+	for _, blobber := range a.Blobbers {
+		addLogChan(blobber.Baseurl)
+	}
 	a.generateAndSetOwnerSigningPublicKey()
 	a.startWorker(a.ctx)
 	InitCommitWorker(a.Blobbers)
@@ -3406,3 +3411,38 @@ func contextCanceled(ctx context.Context) bool {
 		return false
 	}
 }
+
+type logEntry struct {
+	OpType    string
+	DataSize  int
+	TimeTaken int64
+}
+
+func addLogChan(blobberURL string) {
+	logMapMutex.Lock()
+	defer logMapMutex.Unlock()
+	if _, ok := logChanMap[blobberURL]; ok {
+		return
+	}
+	logChan := make(chan logEntry, 200)
+	logChanMap[blobberURL] = logChan
+	go logWorker(blobberURL, logChan)
+}
+
+func getLogChan(blobberURL string) chan logEntry {
+	logMapMutex.Lock()
+	defer logMapMutex.Unlock()
+	return logChanMap[blobberURL]
+}
+
+func logWorker(key string, logChan chan logEntry) {
+	for log := range logChan {
+		data, _ := json.Marshal(log)
+		sys.Files.StoreLogs(key, string(data))
+	}
+}
+
+func writeLogEntry(blobberURL string, log logEntry) {
+	logChan := getLogChan(blobberURL)
+	logChan <- log
+}

zboxcore/sdk/blockdownloadworker.go

@@ -188,7 +188,12 @@ func (req *BlockDownloadRequest) downloadBlobberBlock(fastClient *fasthttp.Clien
 				}
 				return errors.New("response_error", string(respBuf))
 			}
-
+			entry := logEntry{
+				OpType:    "download",
+				TimeTaken: timeTaken,
+				DataSize:  len(req.respBuf),
+			}
+			writeLogEntry(req.blobber.Baseurl, entry)
 			dR := downloadResponse{}
 			if req.shouldVerify {
 				err = json.Unmarshal(respBuf, &dR)