Add split phase metrics for filestream fsync operations (#10493)

## Summary

This PR implements split phase metrics for filestream operations as
requested in #10432.

### Changes

- Added `vm_filestream_fsync_duration_seconds_total` metric to track
fsync syscall duration separately
- Added `vm_filestream_fsync_calls_total` metric to count fsync calls
- Added `vm_filestream_write_syscall_duration_seconds_total` metric to
track write syscall duration (previously mixed with flush time)
- Refactored `MustClose()` and `MustFlush()` to use new `flush()` and
`sync()` helper methods
- Kept `vm_filestream_write_duration_seconds_total` for backward
compatibility

### Problem Solved

Previously, `vm_filestream_write_duration_seconds_total` was being
incremented in two places:
1. `statWriter.Write()` - triggered by `bw.Flush()` and `bw.Write()`
2. `Writer.MustFlush()` - which included the above process, leading to
double-counting

This made it impossible to distinguish between write syscall time and
fsync time, which is critical for diagnosing storage latency issues.

### Solution

The new metrics allow users to:
- Distinguish "flush got slower" vs "fsync got slower" using metrics
only
- No file path labels (bounded cardinality)
- No double-counting between metrics

### Testing

- Code compiles successfully
- All existing metrics are preserved for backward compatibility

Closes #10432

---------

Signed-off-by: Aliaksandr Valialkin <valyala@victoriametrics.com>
Signed-off-by: Aliaksandr Valialkin <valyala@gmail.com>
Co-authored-by: Aliaksandr Valialkin <valyala@victoriametrics.com>
Co-authored-by: Aliaksandr Valialkin <valyala@gmail.com>

lib/filestream/filestream.go

@@ -243,17 +243,12 @@ func newWriter(f *os.File, nocache bool) *Writer {
 
 // MustClose syncs the underlying file to storage and then closes it.
 func (w *Writer) MustClose() {
-	if err := w.bw.Flush(); err != nil {
-		logger.Panicf("FATAL: cannot flush buffered data to file %q: %s", w.f.Name(), err)
-	}
+	w.flush()
+
 	putBufioWriter(w.bw)
 	w.bw = nil
 
-	if !fsutil.IsFsyncDisabled() {
-		if err := w.f.Sync(); err != nil {
-			logger.Panicf("FATAL: cannot sync file %q: %s", w.f.Name(), err)
-		}
-	}
+	w.sync()
 	if err := w.st.close(); err != nil {
 		logger.Panicf("FATAL: cannot close streamTracker for file %q: %s", w.f.Name(), err)
 	}
@@ -265,6 +260,24 @@ func (w *Writer) MustClose() {
 	writersCount.Dec()
 }
 
+func (w *Writer) flush() {
+	if err := w.bw.Flush(); err != nil {
+		logger.Panicf("FATAL: cannot flush buffered data to file %q: %s", w.f.Name(), err)
+	}
+}
+
+func (w *Writer) sync() {
+	if !fsutil.IsFsyncDisabled() {
+		startTime := time.Now()
+		if err := w.f.Sync(); err != nil {
+			logger.Panicf("FATAL: cannot sync file %q: %s", w.f.Name(), err)
+		}
+		d := time.Since(startTime).Seconds()
+		fsyncDuration.Add(d)
+		fsyncCalls.Inc()
+	}
+}
+
 var (
 	writeDuration        = metrics.NewFloatCounter(`vm_filestream_write_duration_seconds_total`)
 	writeCallsBuffered   = metrics.NewCounter(`vm_filestream_buffered_write_calls_total`)
@@ -272,6 +285,9 @@ var (
 	writtenBytesBuffered = metrics.NewCounter(`vm_filestream_buffered_written_bytes_total`)
 	writtenBytesReal     = metrics.NewCounter(`vm_filestream_real_written_bytes_total`)
 	writersCount         = metrics.NewCounter(`vm_filestream_writers`)
+
+	fsyncDuration        = metrics.NewFloatCounter(`vm_filestream_fsync_duration_seconds_total`)
+	fsyncCalls           = metrics.NewCounter(`vm_filestream_fsync_calls_total`)
 )
 
 // Write writes p to the underlying file.
@@ -292,13 +308,9 @@ func (w *Writer) Write(p []byte) (int, error) {
 //
 // if isSync is true, then the flushed data is fsynced to the underlying storage.
 func (w *Writer) MustFlush(isSync bool) {
-	if err := w.bw.Flush(); err != nil {
-		logger.Panicf("FATAL: cannot flush buffered data to file %q: %s", w.f.Name(), err)
-	}
+	w.flush()
 	if isSync {
-		if err := w.f.Sync(); err != nil {
-			logger.Panicf("FATAL: cannot fsync data to the underlying storage for file %q: %s", w.f.Name(), err)
-		}
+		w.sync()
 	}
 }