python - How to get ordered logs using GCP PubSub - Stack Overflow

admin管理员组

文章数量:1277902

I am using the GCP Pub/Sub to get real-time google logs into my Python application.

I have created a log Sink that routes specific cloud run job logs into a Pub/Sub topic and I am subscribing to that topic from my Python application. Additionally, I have enabled the "Message ordering" option when creating the Pub/Sub subscription -- but I am pretty sure that doesn't do what I think it does.

The flow looks like this:

1. From my Python application I create and run a google cloud run job
2. The logs from that run job are routed to the Pub/Sub topic
3. I subscribe to the Pub/Sub subscription which is linked to the topic to get real-time logs for that specific job

This all works, however, the logs that are coming from that topic are not coming in the same order they are created on Google.

Here is the relevant log streaming code from my python app:

def _log_callback(log_entry):
    print(f"{log_entry['timestamp']}: {log_entry['textPayload']}")


class GoogleSubscriberClient:
    def __init__(self, subscription_id: str):
        storage_credentials = get_google_credentials()
        self.client = pubsub_v1.SubscriberClient(credentials=storage_credentials)

        self.subscription_id = subscription_id
        self.project_id = os.environ.get("GCLOUD_PROJECT_ID")

        self.subscription_path = self.client.subscription_path(
            self.project_id, self.subscription_id
        )

        self._stop_event = threading.Event()

    def process_message(self, message, job_name):
        try:
            data = json.loads(message.data.decode("utf-8"))

            resource = data.get("resource", {})
            labels = resource.get("labels", {})
            log_job_name = labels.get("job_name", "")

            if log_job_name == job_name:
                _log_callback(data)

            message.ack()
        except json.JSONDecodeError:
            print(f"Failed to decode message: {message.data}")
            message.ack()
        except Exception as e:
            print(f"Error processing message: {e}")
            message.ack()

    def listen_for_logs(self, job_name, timeout=30):
        def callback_wrapper(message):
            self.process_message(message, job_name)

        streaming_pull = self.client.subscribe(
            subscription=self.subscription_path, callback=callback_wrapper
        )

        with self.client:
            try:
                streaming_pull.result(timeout=timeout)
            except TimeoutError:
                streaming_pull.cancel()
                streaming_pull.result()
            finally:
                self._stop_event.clear()

    def stop(self):
        self._stop_event.set()

Does anyone have any idea how I would resolve this, real time logs that are ordered by the timestamp are really important in this case.

Thank you.

本文标签： pythonHow to get ordered logs using GCP PubSubStack Overflow