chore(19408): track PcesWriter metrics more granularly (hiero-ledger#19458)

Signed-off-by: mxtartaglia <maxi@swirldslabs.com>

Author: mxtartaglia-sl

platform-sdk/base-utility/src/main/java/org/hiero/base/io/streams/SerializableDataOutputStream.java

@@ -254,9 +254,12 @@ protected void writeClassIdVersion(@NonNull final SerializableDet serializable,
      * @param codec the codec to use to write the record
      * @param <T> the type of the record
      * @throws IOException thrown if any IO problems occur
+     * @return the length in bytes that were written
      */
-    public <T> void writePbjRecord(@NonNull final T record, @NonNull final Codec<T> codec) throws IOException {
-        writeInt(codec.measureRecord(record));
+    public <T> long writePbjRecord(@NonNull final T record, @NonNull final Codec<T> codec) throws IOException {
+        final int recordSize = codec.measureRecord(record);
+        writeInt(recordSize);
         codec.write(record, writableSequentialData);
+        return recordSize + Integer.BYTES;
     }
 }

platform-sdk/swirlds-platform-core/src/main/java/com/swirlds/platform/event/preconsensus/DefaultInlinePcesWriter.java

@@ -11,9 +11,11 @@
 import org.hiero.consensus.model.node.NodeId;
 
 public class DefaultInlinePcesWriter implements InlinePcesWriter {
+
     private final CommonPcesWriter commonPcesWriter;
     private final NodeId selfId;
     private final FileSyncOption fileSyncOption;
+    private final PcesWriterPerEventMetrics pcesWriterPerEventMetrics;
 
     /**
      * Constructor
@@ -33,6 +35,9 @@ public DefaultInlinePcesWriter(
                 .getConfiguration()
                 .getConfigData(PcesConfig.class)
                 .inlinePcesSyncOption();
+
+        this.pcesWriterPerEventMetrics =
+                new PcesWriterPerEventMetrics(platformContext.getMetrics(), platformContext.getTime());
     }
 
     @Override
@@ -45,7 +50,9 @@ public void beginStreamingNewEvents() {
      */
     @NonNull
     @Override
-    public PlatformEvent writeEvent(@NonNull PlatformEvent event) {
+    public PlatformEvent writeEvent(@NonNull final PlatformEvent event) {
+        pcesWriterPerEventMetrics.startWriteEvent();
+
         // if we aren't streaming new events yet, assume that the given event is already durable
         if (!commonPcesWriter.isStreamingNewEvents()) {
             return event;
@@ -58,17 +65,24 @@ public PlatformEvent writeEvent(@NonNull PlatformEvent event) {
 
         try {
             commonPcesWriter.prepareOutputStream(event);
-            commonPcesWriter.getCurrentMutableFile().writeEvent(event);
+            pcesWriterPerEventMetrics.startFileWrite();
+            final long size = commonPcesWriter.getCurrentMutableFile().writeEvent(event);
+            pcesWriterPerEventMetrics.endFileWrite(size);
 
             if (fileSyncOption == FileSyncOption.EVERY_EVENT
                     || (fileSyncOption == FileSyncOption.EVERY_SELF_EVENT
                             && event.getCreatorId().equals(selfId))) {
+
+                pcesWriterPerEventMetrics.startFileSync();
                 commonPcesWriter.getCurrentMutableFile().sync();
+                pcesWriterPerEventMetrics.endFileSync();
             }
-
             return event;
         } catch (final IOException e) {
             throw new UncheckedIOException(e);
+        } finally {
+            pcesWriterPerEventMetrics.endWriteEvent();
+            pcesWriterPerEventMetrics.clear();
         }
     }
 

platform-sdk/swirlds-platform-core/src/main/java/com/swirlds/platform/event/preconsensus/PcesFileChannelWriter.java

@@ -29,9 +29,6 @@ public class PcesFileChannelWriter implements PcesFileWriter {
     private WritableSequentialData writableSequentialData;
     /** Tracks the size of the file in bytes */
     private int fileSize;
-    /** Keeps stats of the writing process */
-    private final PcesFileWriterStats stats;
-
     /**
      * Create a new writer that writes events to a file using a {@link FileChannel}.
      *
@@ -57,7 +54,6 @@ public PcesFileChannelWriter(@NonNull final Path filePath, @NonNull final List<O
         this.channel = FileChannel.open(filePath, allOpenOptions.toArray(OpenOption[]::new));
         this.buffer = ByteBuffer.allocateDirect(BUFFER_CAPACITY);
         this.writableSequentialData = BufferedData.wrap(buffer);
-        this.stats = new PcesFileWriterStats();
     }
 
     @Override
@@ -67,23 +63,18 @@ public void writeVersion(final int version) throws IOException {
     }
 
     @Override
-    public void writeEvent(@NonNull final GossipEvent event) throws IOException {
-        long startTime = System.nanoTime();
+    public long writeEvent(@NonNull final GossipEvent event) throws IOException {
         final int size = GossipEvent.PROTOBUF.measureRecord(event);
-        boolean bufferExpanded = false;
-        try {
-            if (size > buffer.capacity()) {
-                MemoryUtils.closeDirectByteBuffer(buffer);
-                buffer = ByteBuffer.allocateDirect(size);
-                writableSequentialData = BufferedData.wrap(buffer);
-                bufferExpanded = true;
-            }
-            buffer.putInt(size);
-            GossipEvent.PROTOBUF.write(event, writableSequentialData);
-            flipWriteClear();
-        } finally {
-            stats.updateWriteStats(startTime, System.nanoTime(), size, bufferExpanded);
+        final boolean expandBuffer = size > buffer.capacity();
+        if (expandBuffer) {
+            MemoryUtils.closeDirectByteBuffer(buffer);
+            buffer = ByteBuffer.allocateDirect(size);
+            writableSequentialData = BufferedData.wrap(buffer);
         }
+        buffer.putInt(size);
+        GossipEvent.PROTOBUF.write(event, writableSequentialData);
+        flipWriteClear();
+        return size;
     }
 
     /**
@@ -94,18 +85,13 @@ public void writeEvent(@NonNull final GossipEvent event) throws IOException {
     private void flipWriteClear() throws IOException {
 
         buffer.flip();
-        long writeStart = System.nanoTime();
-        try {
-            final int bytesWritten = channel.write(buffer);
-            fileSize += bytesWritten;
-            if (bytesWritten != buffer.limit()) {
-                throw new IOException(
-                        "Failed to write data to file. Wrote " + bytesWritten + " bytes out of " + buffer.limit());
-            }
-            buffer.clear();
-        } finally {
-            stats.updatePartialWriteStats(writeStart, System.nanoTime());
+        final int bytesWritten = channel.write(buffer);
+        fileSize += bytesWritten;
+        if (bytesWritten != buffer.limit()) {
+            throw new IOException(
+                    "Failed to write data to file. Wrote " + bytesWritten + " bytes out of " + buffer.limit());
         }
+        buffer.clear();
     }
 
     @Override
@@ -115,11 +101,8 @@ public void flush() throws IOException {
 
     @Override
     public void sync() throws IOException {
-        long startTime = System.nanoTime();
         // benchmarks show that this has horrible performance for the channel writer (in mac-os)
         channel.force(false);
-
-        stats.updateSyncStats(startTime, System.nanoTime());
     }
 
     @Override
@@ -131,8 +114,4 @@ public void close() throws IOException {
     public long fileSize() {
         return fileSize;
     }
-
-    public PcesFileWriterStats getStats() {
-        return stats;
-    }
 }

platform-sdk/swirlds-platform-core/src/main/java/com/swirlds/platform/event/preconsensus/PcesFileManager.java

@@ -237,7 +237,6 @@ public void finishedWritingFile(@NonNull final PcesMutableFile file) {
         metrics.getPreconsensusEventAverageFileSpan().update(file.getSpan());
         metrics.getPreconsensusEventAverageUnUtilizedFileSpan().update(file.getUnUtilizedSpan());
         updateFileSizeMetrics();
-        metrics.updateMetricsWithPcesFileWritingStats(file.writerStats());
     }
 
     /**

platform-sdk/swirlds-platform-core/src/main/java/com/swirlds/platform/event/preconsensus/PcesFileWriter.java

@@ -21,8 +21,9 @@ public interface PcesFileWriter {
      * Write an event to the file.
      *
      * @param event the event to write
+     * @return the length of the written data
      */
-    void writeEvent(@NonNull final GossipEvent event) throws IOException;
+    long writeEvent(@NonNull final GossipEvent event) throws IOException;
 
     /**
      * Flush the file.
@@ -43,7 +44,4 @@ public interface PcesFileWriter {
      * @return the size of the file in bytes
      */
     long fileSize();
-
-    /** Keeps stats of the writing process */
-    PcesFileWriterStats getStats();
 }

platform-sdk/swirlds-platform-core/src/main/java/com/swirlds/platform/event/preconsensus/PcesFileWriterStats.java

@@ -3,20 +3,17 @@
 
 import com.swirlds.common.metrics.RunningAverageMetric;
 import com.swirlds.common.metrics.SpeedometerMetric;
-import com.swirlds.metrics.api.Counter;
 import com.swirlds.metrics.api.DoubleGauge;
 import com.swirlds.metrics.api.LongGauge;
 import com.swirlds.metrics.api.Metrics;
 import edu.umd.cs.findbugs.annotations.NonNull;
-import java.util.Objects;
 
 /**
  * Metrics for preconsensus events.
  */
 public class PcesMetrics {
 
     private static final String CATEGORY = "platform";
-    private final Metrics metrics;
 
     private static final LongGauge.Config PRECONSENSUS_EVENT_FILE_COUNT_CONFIG = new LongGauge.Config(
                     CATEGORY, "preconsensusEventFileCount")
@@ -72,29 +69,12 @@ public class PcesMetrics {
             .withDescription("The age of the oldest preconsensus event file, in seconds.");
     private final LongGauge preconsensusEventFileOldestSeconds;
 
-    private static final RunningAverageMetric.Config AVG_EVENT_SIZE = new RunningAverageMetric.Config(
-                    CATEGORY, "pcesAvgEventSize")
-            .withDescription("The average size of an event");
-    private static final RunningAverageMetric.Config PCES_AVG_SYNC_DURATION = new RunningAverageMetric.Config(
-                    CATEGORY, "pcesAvgSyncDuration")
-            .withDescription("The average duration of the sync method");
-    private static final RunningAverageMetric.Config PCES_AVG_WRITE_DURATION = new RunningAverageMetric.Config(
-                    CATEGORY, "pcesAvgWriteDuration")
-            .withDescription("The average duration of the write fs call");
-    private static final RunningAverageMetric.Config PCES_AVG_TOTAL_WRITE_DURATION = new RunningAverageMetric.Config(
-                    CATEGORY, "pcesAvgTotalWriteDuration")
-            .withDescription("The average of the total duration of the write method");
-    private static final Counter.Config PCES_BUFFER_EXPANSIONS_COUNTER = new Counter.Config(
-                    CATEGORY, "pcesBufferExpansionCounter")
-            .withDescription("How many times the write buffer needed to be expanded");
-
     /**
      * Construct preconsensus event metrics.
      *
      * @param metrics the metrics manager for the platform
      */
     public PcesMetrics(final @NonNull Metrics metrics) {
-        this.metrics = Objects.requireNonNull(metrics);
         preconsensusEventFileCount = metrics.getOrCreate(PRECONSENSUS_EVENT_FILE_COUNT_CONFIG);
         preconsensusEventFileAverageSizeMB = metrics.getOrCreate(PRECONSENSUS_EVENT_FILE_AVERAGE_SIZE_MB_CONFIG);
         preconsensusEventFileTotalSizeGB = metrics.getOrCreate(PRECONSENSUS_EVENT_FILE_TOTAL_SIZE_GB_CONFIG);
@@ -106,13 +86,6 @@ public PcesMetrics(final @NonNull Metrics metrics) {
         preconsensusEventFileYoungestIdentifier =
                 metrics.getOrCreate(PRECONSENSUS_EVENT_FILE_YOUNGEST_IDENTIFIER_CONFIG);
         preconsensusEventFileOldestSeconds = metrics.getOrCreate(PRECONSENSUS_EVENT_FILE_OLDEST_SECONDS_CONFIG);
-
-        // crating the metrics
-        metrics.getOrCreate(PcesMetrics.AVG_EVENT_SIZE);
-        metrics.getOrCreate(PcesMetrics.PCES_AVG_SYNC_DURATION);
-        metrics.getOrCreate(PcesMetrics.PCES_AVG_WRITE_DURATION);
-        metrics.getOrCreate(PcesMetrics.PCES_AVG_TOTAL_WRITE_DURATION);
-        metrics.getOrCreate(PcesMetrics.PCES_BUFFER_EXPANSIONS_COUNTER);
     }
 
     /**
@@ -177,16 +150,4 @@ public LongGauge getPreconsensusEventFileYoungestIdentifier() {
     public LongGauge getPreconsensusEventFileOldestSeconds() {
         return preconsensusEventFileOldestSeconds;
     }
-
-    /**
-     * Updates the metrics with the stats reported by the writer
-     */
-    public void updateMetricsWithPcesFileWritingStats(@NonNull final PcesFileWriterStats stats) {
-        metrics.getOrCreate(PcesMetrics.AVG_EVENT_SIZE).update(stats.averageEventSize());
-        metrics.getOrCreate(PcesMetrics.PCES_AVG_SYNC_DURATION).update(stats.averageSyncDuration());
-        metrics.getOrCreate(PcesMetrics.PCES_AVG_WRITE_DURATION).update(stats.averageWriteDuration());
-        metrics.getOrCreate(PcesMetrics.PCES_AVG_TOTAL_WRITE_DURATION).update(stats.averageTotalWriteDuration());
-        if (stats.totalExpansions() > 0)
-            metrics.getOrCreate(PcesMetrics.PCES_BUFFER_EXPANSIONS_COUNTER).add(stats.totalExpansions());
-    }
 }