[#159] Add new pool metric pgagroal_tx_count

contrib/grafana/dashboard.json

@@ -653,6 +653,85 @@
       "title": "Query rate",
       "type": "timeseries"
     },
+    {
+      "datasource": null,
+      "description": "",
+      "fieldConfig": {
+        "defaults": {
+          "color": {
+            "mode": "palette-classic"
+          },
+          "custom": {
+            "axisLabel": "",
+            "axisPlacement": "auto",
+            "barAlignment": 0,
+            "drawStyle": "line",
+            "fillOpacity": 0,
+            "gradientMode": "none",
+            "hideFrom": {
+              "legend": false,
+              "tooltip": false,
+              "viz": false
+            },
+            "lineInterpolation": "linear",
+            "lineWidth": 1,
+            "pointSize": 5,
+            "scaleDistribution": {
+              "type": "linear"
+            },
+            "showPoints": "auto",
+            "spanNulls": false,
+            "stacking": {
+              "group": "A",
+              "mode": "none"
+            },
+            "thresholdsStyle": {
+              "mode": "off"
+            }
+          },
+          "mappings": [],
+          "thresholds": {
+            "mode": "absolute",
+            "steps": [
+              {
+                "color": "green",
+                "value": null
+              }
+            ]
+          },
+          "unit": "percent"
+        },
+        "overrides": []
+      },
+      "gridPos": {
+        "h": 8,
+        "w": 11,
+        "x": 11,
+        "y": 26
+      },
+      "id": 44,
+      "options": {
+        "legend": {
+          "calcs": [],
+          "displayMode": "list",
+          "placement": "bottom"
+        },
+        "tooltip": {
+          "mode": "single"
+        }
+      },
+      "targets": [
+        {
+          "exemplar": true,
+          "expr": "rate(pgagroal_tx_count[5m])",
+          "interval": "",
+          "legendFormat": "",
+          "refId": "A"
+        }
+      ],
+      "title": "Transaction Rate",
+      "type": "timeseries"
+    },
     {
       "datasource": null,
       "gridPos": {
@@ -1121,5 +1200,5 @@
   "timezone": "",
   "title": "pgagroal dashboard",
   "uid": "t_a1YcR7k",
-  "version": 27
+  "version": 28
 }

src/include/pgagroal.h

@@ -252,6 +252,7 @@ struct prometheus
    atomic_ulong client_wait_time;       /**< The time the client waits */
 
    atomic_ullong query_count;           /**< The number of queries */
+   atomic_ullong tx_count;              /**< The number of transactions */
 
    atomic_ulong server_error[NUMBER_OF_SERVERS]; /**< The number of errors for a server */
    atomic_ulong failed_servers;                  /**< The number of failed servers */

src/include/prometheus.h

@@ -164,6 +164,12 @@ pgagroal_prometheus_client_active_sub(void);
 void
 pgagroal_prometheus_query_count_add(void);
 
+/**
+ * Increase tx_count by 1
+ */
+void
+pgagroal_prometheus_tx_count_add(void);
+
 /**
  * Reset the counters and histograms
  */

src/libpgagroal/pipeline_session.c

@@ -35,6 +35,7 @@
 #include <server.h>
 #include <shmem.h>
 #include <worker.h>
+#include <utils.h>
 
 /* system */
 #include <errno.h>
@@ -49,6 +50,9 @@ static void session_stop(struct ev_loop *loop, struct worker_io*);
 static void session_destroy(void*, size_t);
 static void session_periodic(void);
 
+static bool in_tx;
+static int next_server_message;
+
 #define CLIENT_INIT   0
 #define CLIENT_IDLE   1
 #define CLIENT_ACTIVE 2
@@ -120,6 +124,9 @@ session_start(struct ev_loop *loop, struct worker_io* w)
 {
    struct client_session* client;
 
+   in_tx = false;
+   next_server_message = 0;
+
    if (pipeline_shmem != NULL)
    {
       client = pipeline_shmem + (w->slot * sizeof(struct client_session));
@@ -307,6 +314,46 @@ session_server(struct ev_loop *loop, struct ev_io *watcher, int revents)
    status = pgagroal_read_socket_message(wi->server_fd, &msg);
    if (likely(status == MESSAGE_STATUS_OK))
    {
+      int offset = 0;
+
+      while (offset < msg->length)
+      {
+         if (next_server_message == 0)
+         {
+            char kind = pgagroal_read_byte(msg->data + offset);
+            int length = pgagroal_read_int32(msg->data + offset + 1);
+
+            /* The Z message tell us the transaction state */
+            if (kind == 'Z')
+            {
+               char tx_state = pgagroal_read_byte(msg->data + offset + 5);
+
+               if (tx_state != 'I' && !in_tx)
+               {
+                  pgagroal_prometheus_tx_count_add();
+               }
+
+               in_tx = tx_state != 'I';
+            }
+
+            /* Calculate the offset to the next message */
+            if (offset + length + 1 <= msg->length)
+            {
+               next_server_message = 0;
+               offset += length + 1;
+            }
+            else
+            {
+               next_server_message = length + 1 - (msg->length - offset);
+               offset = msg->length;
+            }
+         }
+         else
+         {
+            offset = MIN(next_server_message, msg->length);
+            next_server_message -= offset;
+         }
+      }
       if (wi->client_ssl == NULL)
       {
          status = pgagroal_write_socket_message(wi->client_fd, msg);

src/libpgagroal/pipeline_transaction.c

@@ -387,14 +387,12 @@ transaction_server(struct ev_loop *loop, struct ev_io *watcher, int revents)
 
                has_z = true;
 
-               if (tx_state == 'I')
+               if (tx_state != 'I' && !in_tx)
                {
-                  in_tx = false;
-               }
-               else
-               {
-                  in_tx = true;
+                  pgagroal_prometheus_tx_count_add();
                }
+
+               in_tx = tx_state != 'I';
             }
 
             /* Calculate the offset to the next message */

src/libpgagroal/prometheus.c

@@ -186,6 +186,7 @@ pgagroal_init_prometheus(size_t* p_size, void** p_shmem)
    atomic_init(&prometheus->client_wait_time, 0);
 
    atomic_init(&prometheus->query_count, 0);
+   atomic_init(&prometheus->tx_count, 0);
 
    for (int i = 0; i < NUMBER_OF_SERVERS; i++)
    {
@@ -475,6 +476,16 @@ pgagroal_prometheus_query_count_add(void)
    atomic_fetch_add(&prometheus->query_count, 1);
 }
 
+void
+pgagroal_prometheus_tx_count_add(void)
+{
+   struct prometheus* prometheus;
+
+   prometheus = (struct prometheus*)prometheus_shmem;
+
+   atomic_fetch_add(&prometheus->tx_count, 1);
+}
+
 void
 pgagroal_prometheus_reset(void)
 {
@@ -508,6 +519,7 @@ pgagroal_prometheus_reset(void)
    atomic_store(&prometheus->client_wait_time, 0);
 
    atomic_store(&prometheus->query_count, 0);
+   atomic_store(&prometheus->tx_count, 0);
 
    for (int i = 0; i < NUMBER_OF_SERVERS; i++)
    {
@@ -713,6 +725,9 @@ home_page(int client_fd)
    data = append(data, "  <h2>pgagroal_query_count</h2>\n");
    data = append(data, "  The number of queries\n");
    data = append(data, "  <p>\n");
+   data = append(data, "  <h2>pgagroal_tx_count</h2>\n");
+   data = append(data, "  The number of transactions\n");
+   data = append(data, "  <p>\n");
    data = append(data, "  <h2>pgagroal_active_connections</h2>\n");
    data = append(data, "  The number of active connections\n");
    data = append(data, "  <p>\n");
@@ -1016,6 +1031,12 @@ general_information(int client_fd)
    data = append_ullong(data, atomic_load(&prometheus->query_count));
    data = append(data, "\n\n");
 
+   data = append(data, "#HELP pgagroal_tx_count The number of transactions\n");
+   data = append(data, "#TYPE pgagroal_tx_count count\n");
+   data = append(data, "pgagroal_tx_count ");
+   data = append_ullong(data, atomic_load(&prometheus->tx_count));
+   data = append(data, "\n\n");
+
    if (data != NULL)
    {
       send_chunk(client_fd, data);