CNDB-15716: register sensors in paxos v2 (#2070)

### What is the issue

Paxos v2 doesn't have sensors

### What does this PR fix and why was it fixed

Register sensors in paxos v2

Author: driftx

src/java/org/apache/cassandra/service/paxos/PaxosPrepare.java

@@ -52,6 +52,12 @@
 import org.apache.cassandra.schema.Schema;
 import org.apache.cassandra.schema.TableId;
 import org.apache.cassandra.schema.TableMetadata;
+import org.apache.cassandra.sensors.Context;
+import org.apache.cassandra.sensors.RequestSensors;
+import org.apache.cassandra.sensors.RequestTracker;
+import org.apache.cassandra.sensors.SensorsCustomParams;
+import org.apache.cassandra.sensors.SensorsFactory;
+import org.apache.cassandra.sensors.Type;
 import org.apache.cassandra.service.PendingRangeCalculatorService;
 import org.apache.cassandra.service.paxos.PaxosPrepare.Status.Outcome;
 import org.apache.cassandra.tracing.Tracing;
@@ -1021,11 +1027,33 @@ public static class RequestHandler implements IVerbHandler<Request>
         @Override
         public void doVerb(Message<Request> message)
         {
+            // Initialize the sensor and set ExecutorLocals
+            RequestSensors sensors = SensorsFactory.instance.createRequestSensors(message.payload.table.keyspace);
+            Context context = Context.from(message.payload.table);
+
+            // Prepare phase incorporates a read to check the cas condition, so a read sensor is registered in addition to the write sensor
+            sensors.registerSensor(context, Type.READ_BYTES);
+            sensors.registerSensor(context, Type.WRITE_BYTES);
+            sensors.registerSensor(context, Type.INTERNODE_BYTES);
+            sensors.incrementSensor(context, Type.INTERNODE_BYTES, message.payloadSize(MessagingService.current_version));
+            RequestTracker.instance.set(sensors);
+
             Response response = execute(message.payload, message.from());
-            if (response == null)
-                MessagingService.instance().respondWithFailure(UNKNOWN, message);
+
+            // calculate outbound internode bytes before adding the sensor to the response
+            if (response != null)
+            {
+                Message.Builder<Response> reply = message.responseWithBuilder(response);
+                int size = reply.currentPayloadSize(MessagingService.current_version);
+                sensors.incrementSensor(context, Type.INTERNODE_BYTES, size);
+                sensors.syncAllSensors();
+                SensorsCustomParams.addSensorsToInternodeResponse(sensors, reply);
+                MessagingService.instance().send(reply.build(), message.from());
+            }
             else
-                MessagingService.instance().respond(response, message);
+            {
+                MessagingService.instance().respondWithFailure(UNKNOWN, message);
+            }
         }
 
         static Response execute(AbstractRequest<?> request, InetAddressAndPort from)

src/java/org/apache/cassandra/service/paxos/PaxosPropose.java

@@ -37,6 +37,12 @@
 import org.apache.cassandra.net.IVerbHandler;
 import org.apache.cassandra.net.Message;
 import org.apache.cassandra.net.MessagingService;
+import org.apache.cassandra.sensors.Context;
+import org.apache.cassandra.sensors.RequestSensors;
+import org.apache.cassandra.sensors.RequestTracker;
+import org.apache.cassandra.sensors.SensorsCustomParams;
+import org.apache.cassandra.sensors.SensorsFactory;
+import org.apache.cassandra.sensors.Type;
 import org.apache.cassandra.service.paxos.Commit.Proposal;
 import org.apache.cassandra.utils.concurrent.ConditionAsConsumer;
 
@@ -408,11 +414,33 @@ public static class RequestHandler implements IVerbHandler<Request>
         @Override
         public void doVerb(Message<Request> message)
         {
+            // Initialize the sensor and set ExecutorLocals
+            RequestSensors sensors = SensorsFactory.instance.createRequestSensors(message.payload.proposal.update.metadata().keyspace);
+            Context context = Context.from(message.payload.proposal.update.metadata());
+
+            // Propose phase consults the Paxos table for more recent promises, so a read sensor is registered in addition to the write sensor
+            sensors.registerSensor(context, Type.READ_BYTES);
+            sensors.registerSensor(context, Type.WRITE_BYTES);
+            sensors.registerSensor(context, Type.INTERNODE_BYTES);
+            sensors.incrementSensor(context, Type.INTERNODE_BYTES, message.payloadSize(MessagingService.current_version));
+            RequestTracker.instance.set(sensors);
+
             Response response = execute(message.payload.proposal, message.from());
-            if (response == null)
-                MessagingService.instance().respondWithFailure(UNKNOWN, message);
+
+            // calculate outbound internode bytes before adding the sensor to the response
+            if (response != null)
+            {
+                Message.Builder<Response> reply = message.responseWithBuilder(response);
+                int size = reply.currentPayloadSize(MessagingService.current_version);
+                sensors.incrementSensor(context, Type.INTERNODE_BYTES, size);
+                sensors.syncAllSensors();
+                SensorsCustomParams.addSensorsToInternodeResponse(sensors, reply);
+                MessagingService.instance().send(reply.build(), message.from());
+            }
             else
-                MessagingService.instance().respond(response, message);
+            {
+                MessagingService.instance().respondWithFailure(UNKNOWN, message);
+            }
         }
 
         public static Response execute(Proposal proposal, InetAddressAndPort from)