Reputation: 100
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:
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.
Upvotes: 0
Views: 29