How to count messages from a subscription utilizing streaming pull with GCP Pub/Sub?

Question

I am confused about the GCP Pub/Sub REST API.

Background: I am trying to count the number of messages in a pubsub subscription but I can not iterate through the message object streaming pull.

Therefore, I will need to rely upon the REST API provided: https://cloud.google.com/pubsub/docs/pull#asynchronous-pull

Based on my understanding of the REST API: It currently pulls messages as expected but when I try to iterate through a loop the stack trace highlights the Message object cannot be iterable.

What I have tried : with my current implementation, it only repeats 1 message being sent for each company involved

company_name = {}    
if len(message) == 0:
    logging.warning('Nothing pulled from pubsub')
else: 
    logging.info('Pulled %s messages from pubsub' % str(len(message.data)))
    for msg in message:
        if msg.attributes in message:
            agency_name[message.attributes['company_name']] = 1
        else:
            agency_name[message.attributes['company_name']] += 1
message.ack()

what is the best way of achieving this solution?

Answer 1

In addition to what @guillaume said, you can check this GCP Documentation for reading time-series data using Python: https://cloud.google.com/monitoring/docs/samples/monitoring-read-timeseries-simple#code-sample

from google.cloud import monitoring_v3
import time

client = monitoring_v3.MetricServiceClient()
project_name = f"projects/anjela"

now = time.time()
seconds = int(now)
nanos = int((now - seconds) * 10 ** 9)
interval = monitoring_v3.TimeInterval(
    {
        "end_time": {"seconds": seconds, "nanos": nanos},
        "start_time": {"seconds": (seconds - 1200), "nanos": nanos},
    }
)


results = client.list_time_series(
    request={
        "name": project_name,
        "filter": 'metric.type = "pubsub.googleapis.com/subscription/num_undelivered_messages"',
        "interval": interval,
        "view": monitoring_v3.ListTimeSeriesRequest.TimeSeriesView.FULL,
    }
)
for result in results:
    print(result)

I put this filter pubsub.googleapis.com/subscription/num_undelivered_messages as it tracks the number of unacknowledged or backlog messages. You can use this GCP documentation to alter the filter according to your purpose: https://cloud.google.com/monitoring/api/metrics_gcp#gcp-pubsub

Result in Cloud Monitoring Interface:

Response: