add more logs

Author: coanor

diskcache/diskcache.go

@@ -94,7 +94,8 @@ type DiskCache struct {
 	pos   *pos   // current read fd position info
 
 	// specs of current diskcache
-	size          atomic.Int64 // current byte size
+	size atomic.Int64 // current byte size
+
 	curBatchSize, // current writing file's size
 	curReadSize, // current reading file's size
 	batchSize, // current batch size(static)

diskcache/get.go

@@ -82,9 +82,10 @@ func (c *DiskCache) doGet(buf []byte, fn Fn, bfn BufFunc) error {
 		if err = func() error {
 			c.wlock.Lock()
 			defer c.wlock.Unlock()
+
 			return c.rotate()
 		}(); err != nil {
-			return err
+			return fmt.Errorf("wakeup error: %w", err)
 		}
 	}
 
@@ -140,14 +141,15 @@ retry:
 
 	if len(readbuf) < nbytes {
 		// seek to next read position
-		if _, err := c.rfd.Seek(int64(nbytes), io.SeekCurrent); err != nil {
+		if x, err := c.rfd.Seek(int64(nbytes), io.SeekCurrent); err != nil {
 			return fmt.Errorf("rfd.Seek(%d): %w", nbytes, err)
-		}
+		} else {
+			l.Warnf("got %d bytes to buffer with len %d, seek to new read position %d, drop %d bytes within file %s",
+				nbytes, len(readbuf), x, nbytes, c.curReadfile)
 
-		l.Warnf("got %d bytes to read into buffer with length %d", nbytes, len(readbuf))
-
-		droppedDataVec.WithLabelValues(c.path, reasonTooSmallReadBuffer).Observe(float64(nbytes))
-		return ErrTooSmallReadBuf
+			droppedDataVec.WithLabelValues(c.path, reasonTooSmallReadBuffer).Observe(float64(nbytes))
+			return ErrTooSmallReadBuf
+		}
 	}
 
 	if n, err := c.rfd.Read(readbuf[:nbytes]); err != nil {

diskcache/metric.go

@@ -181,7 +181,7 @@ func setupMetrics() {
 		prometheus.GaugeOpts{
 			Namespace: ns,
 			Name:      "size",
-			Help:      "Current cache size that waiting to be consumed(get)",
+			Help:      "Current cache size that waiting to be consumed(get). The size include header bytes",
 		},
 		[]string{"path"},
 	)

diskcache/open.go

@@ -142,10 +142,6 @@ func (c *DiskCache) doOpen() error {
 			switch filepath.Base(path) {
 			case ".lock", ".pos": // ignore them
 			case "data": // not rotated writing file, do not count on sizeVec.
-				c.size.Add(fi.Size())
-				// NOTE: c.size not always equal to sizeVec. c.size used to limit
-				// total bytes used for Put(), but sizeVec used to count size that
-				// waiting to be Get().
 			default:
 				c.size.Add(fi.Size())
 				sizeVec.WithLabelValues(c.path).Add(float64(fi.Size()))
@@ -158,6 +154,8 @@ func (c *DiskCache) doOpen() error {
 	}
 
 	sort.Strings(c.dataFiles) // make file-name sorted for FIFO Get()
+	l.Infof("on open loaded %d files", len(c.dataFiles))
+	datafilesVec.WithLabelValues(c.path).Set(float64(len(c.dataFiles)))
 
 	// first get, try load .pos
 	if !c.noPos {

diskcache/rotate.go

@@ -73,7 +73,7 @@ func (c *DiskCache) rotate() error {
 		return fmt.Errorf("rotate on Rename(%q, %q): %w", c.curWriteFile, newfile, err)
 	}
 
-	// new file added, plus it's size to cache size
+	// new file added, add it's size to cache size
 	if fi, err := os.Stat(newfile); err == nil {
 		if fi.Size() > dataHeaderLen {
 			c.size.Add(fi.Size())