Google PubSub - Counting messages in topic

Question

I've looked over the documentation for Google's PubSub, and also tried looking in Google Cloud Monitoring, but couldn't find any means of figuring out what's the queue size in my topics.

Since I plan on using PubSub for analytics, it's important for me to monitor the queue count, so I could scale up/down the subscriber count.

What am I missing?

Answer 1

The metric you want to look at is "undelivered messages." You should be able to set up alerts or charts that monitor this metric in Google Cloud Monitoring under the "Pub/Sub Subscription" resource type. The number of messages that have not yet been acknowledged by subscribers, i.e., queue size, is a per-subscription metric as opposed to a per-topic metric. For info on the metric, see pubsub.googleapis.com/subscription/num_undelivered_messages in the GCP Metrics List (and others for all of the Pub/Sub metrics available).

Answer 2

This might help if you're looking into a programmatic way to achieve this:

from google.cloud import monitoring_v3
from google.cloud.monitoring_v3 import query

project = "my-project"
client = monitoring_v3.MetricServiceClient()
result = query.Query(
         client,
         project,
         'pubsub.googleapis.com/subscription/num_undelivered_messages', 
         minutes=60).as_dataframe()

print(result['pubsub_subscription'][project]['subscription_name'][0])

Answer 3

The answer to your question is "no", there is no feature for PubSub that shows these counts. The way you have to do it is via log event monitoring using Stackdriver (it took me some time to find that out too).

The colloquial answer to this is do the following, step-by-step:

1. Navigate from GCloud Admin Console to: Monitoring

2. This opens a new window with separate Stackdriver console
3. Navigate in Stackdriver: Dashboards > Create Dashboard

4. Click the Add Chart button top-right of dashboard screen

5. In the input box, type num_undelivered_messages and then SAVE

Answer 4

Updated version based on @steeve's answer. (without pandas dependency)

Please note that you have to specify end_time instead of using default utcnow().

import datetime
from google.cloud import monitoring_v3
from google.cloud.monitoring_v3 import query

project = 'my-project'
sub_name = 'my-sub'
client = monitoring_v3.MetricServiceClient()
result = query.Query(
  client,
  project,
  'pubsub.googleapis.com/subscription/num_undelivered_messages',
  end_time=datetime.datetime.now(),
  minutes=1,
  ).select_resources(subscription_id=sub_name)

for content in result:
  print(content.points[0].value.int64_value)

Answer 5

Here is a java version

package com.example.monitoring;

import static com.google.cloud.monitoring.v3.MetricServiceClient.create;
import static com.google.monitoring.v3.ListTimeSeriesRequest.newBuilder;
import static com.google.monitoring.v3.ProjectName.of;
import static com.google.protobuf.util.Timestamps.fromMillis;
import static java.lang.System.currentTimeMillis;

import com.google.monitoring.v3.ListTimeSeriesRequest;
import com.google.monitoring.v3.TimeInterval;

public class ReadMessagesFromGcp {

  public static void main(String... args) throws Exception {
   
    String projectId = "put here";

    var interval = TimeInterval.newBuilder()
                               .setStartTime(fromMillis(currentTimeMillis() - (120 * 1000)))
                               .setEndTime(fromMillis(currentTimeMillis()))
                               .build();

    var request = newBuilder().setName(of(projectId).toString())
           .setFilter("metric.type=\"pubsub.googleapis.com/subscription/num_undelivered_messages\"")
           .setInterval(interval)
           .setView(ListTimeSeriesRequest.TimeSeriesView.FULL)
           .build();

    var response = create().listTimeSeries(request);

    for (var subscriptionData : response.iterateAll()) {
        
        var subscription = subscriptionData.getResource().getLabelsMap().get("subscription_id");
        
        var numberOrMessages = subscriptionData.getPointsList().get(0).getValue().getInt64Value();
            
        if(numberOrMessages > 0) {
            System.out.println(subscription + " has " + numberOrMessages + " messages ");
        }
            
    }
  }
}

<dependency>
            <groupId>com.google.cloud</groupId>
            <artifactId>google-cloud-monitoring</artifactId>
            <version>3.3.2</version>
        </dependency>
    
        <dependency>
          <groupId>com.google.protobuf</groupId>
          <artifactId>protobuf-java-util</artifactId>
          <version>4.0.0-rc-2</version>
        </dependency>

output

queue-1 has 36 messages

queue-2 has 4 messages

queue-3 has 3 messages

Answer 6

There is a way to count all messages published to a topic using custom metrics.

In my case I am publishing messages to a Pub/Sub topic via a Cloud Composer (Airflow) Dag that runs a python script.

The python script returns logging information about the ran Dag.

logging.info(
f"Total events in file {counter-1}, total successfully published {counter - error_counter -1}, total errors publishing {error_counter}. Events sent to topic: {TOPIC_PATH} from filename: {source_blob_name}.",
{
"metric": "<some_name>",
"type": "completed_file",
"topic": EVENT_TOPIC,
"filename": source_blob_name,
"total_events_in_file": counter - 1,
"failed_published_messages": error_counter,
"successful_published_messages": counter - error_counter - 1,
}

I then have a Distribution custom metric which filters on resource_type, resource_lable, jsonPayload.metric and jsonPayload.type. The metric also has the Field Name set to jsonPayload.successful_published_messages

Custom metric filter:

resource.type=cloud_composer_environment AND resource.labels.environment_name={env_name} AND jsonPayload.metric=<some_name> AND jsonPayload.type=completed_file

That custom metric is then used in a Dashboard with the MQL setting of

fetch cloud_composer_environment
| metric
'logging.googleapis.com/user/my_custom_metric'
| group_by 1d, [value_pubsub_aggregate: aggregate(value.pubsub)]
| every 1d
| group_by [],
[value_pubsub_aggregate_sum: sum(value_pubsub_aggregate)]

Which to get to I first setup an Icon chart with resource type: cloud composer environment, Metric: my_custom metric, Processing step: to no preprocessing step, Alignment function: SUM, period 1, unit day, How do you want it grouped group by function: mean.

Ideally you would just select sum for the Group by function but it errors and that is why you then need to sqitch to MQL and manually enter sum instead of mean.

This will now count your published messages for up to 24 months which is the retention period set by Google for the custom metrics.