Programmiererforum

def check_worker_logs(event_uuid, dataflow_project, dataflow_job, timeframe_mins=30):
# Start time of the worker log
start_time = (datetime.utcnow() - timedelta(minutes=timeframe_mins)).isoformat() + "Z"
print(start_time)

# Test connection by listing log entries
try:
client = logging.Client()
except Exception as e:
print(f"Failed to connect to Google Cloud Logging: {e}")

# Define the log filter for all logs in the given timeframe
worker_log_filter = (
f'resource.labels.project_id="{dataflow_project}" AND '
f'resource.labels.job_name="{dataflow_job}" AND '
f'timestamp >= "{start_time}" AND '
f'resource.type="dataflow_step" AND '
f'log_id("dataflow.googleapis.com/worker")'
)

print(worker_log_filter)

# Fetch logs for the success log
worker_logs = client.list_entries(filter_=worker_log_filter)
if not any(worker_logs):
print("No worker logs found.")
< /code>
Ich versuche, Datenflow-Arbeiterprotokolle mit der Python Google-Cloud-Logging-Bibliothek abzurufen. /> [*] Wenn ich jedoch den gleichen Filter in meinem Python-Code verwende, werden keine Protokolle zurückgegeben. (Job_Name, Ressource.Type und Log_id), dann gibt der Code Protokolle meines GCP -Projekts zurück - aber nur Prüfungsprotokolle mit Typtyp.googleapis.com/google.cloud.audit.Auditlog.
So scheinen die API -Logging -Log -Log -Setups zu fällen. /> < /ul>
Hat jemand dieses [url=viewtopic.php?t=26065]Problem[/url] schon einmal gestoßen? Gibt es einen subtilen Unterschied zwischen der Art und Weise, wie der Protokoll -Explorer und die Protokollierungs -API Filter interpretieren?>