Direct Library Usage

We recommend that you use the google-cloud-logging library by integrating it with the Python logging standard library; However, you can also use the library to interact with the Google Cloud Logging API directly.

In addition to writing logs, you can use the library to manage logs, sinks, metrics, and other resources.

Setup

Create a Client

You must set up a Client to use the library:

import google.cloud.logging

# if project not given, it will be inferred from the environment
client = google.cloud.logging.Client(project="my-project")

To use HTTP, disable gRPC when you set up the Client:

http_client = google.cloud.logging.Client(_use_grpc=False)

Create a Logger

Loggers read, write, and delete logs from Google Cloud.

You use your Client to create a Logger.

client = google.cloud.logging.Client(project="my-project")
logger = client.logger(name="log_id")
# logger will bind to logName "projects/my_project/logs/log_id"

To add custom labels, do so when you initialize a Logger. When you add custom labels, these labels are added to each LogEntry written by the Logger:

custom_labels = {"my-key": "my-value"}
label_logger = client.logger(log_id, labels=custom_labels)

By default, the library adds a Monitored Resource field associated with the environment the code is run on. For example, code run on App Engine will have a gae_app resource, while code run locally will have a global resource field.

To manually set the resource field, do so when you initialize the Logger:

from google.cloud.logging_v2.resource import Resource

resource = Resource(type="global", labels={})
global_logger = client.logger(log_id, resource=resource)

Write Log Entries

You write logs by using Logger.log:

logger.log("A simple entry")  # API call

You can add LogEntry fields by passing them as keyword arguments:

logger.log(
    "an entry with fields set",
    severity="ERROR",
    insert_id="0123",
    labels={"my-label": "my-value"},
)  # API call

Logger.log chooses the appropriate LogEntry type based on input type. To specify type, you can use the following Logger methods:

Logger.log_text creates a TextEntry
Logger.log_struct creates a StructEntry
Logger.log_proto creates a ProtobufEntry
Logger.log_empty creates an empty LogEntry

Batch Write Logs

By default, each log write takes place in an individual network request, which may be inefficient at scale.

Using the Batch class, logs are batched together, and only sent out when batch.commit is called.

batch = logger.batch()
batch.log("first log")
batch.log("second log")
batch.commit()

To simplify things, you can also use Batch as a context manager:

with logger.batch() as batch:
    batch.log("first log")
    # do work
    batch.log("last log")

In the previous example, the logs are automatically committed when the code exits the “with” block.

Retrieve Log Entries

You retrieve log entries for the default project using list_entries() on a Client or Logger object:

for entry in client.list_entries():  # API call(s)
    do_something_with(entry)

Entries returned by Client.list_entries() or Logger.list_entries() are instances of one of the following classes:

To filter entries retrieved using the Advanced Logs Filters syntax

To fetch entries for the default project.

filter_str = "logName:log_name AND textPayload:simple"
for entry in client.list_entries(filter_=filter_str):  # API call(s)
    do_something_with(entry)

To sort entries in descending timestamp order.

from google.cloud.logging import DESCENDING

for entry in client.list_entries(order_by=DESCENDING):  # API call(s)
    do_something_with(entry)

To retrieve entries for a single logger, sorting in descending timestamp order:

from google.cloud.logging import DESCENDING

for entry in logger.list_entries(order_by=DESCENDING):  # API call(s)
    do_something_with(entry)

For example, to retrieve all GKE Admin Activity audit logs from the past 24 hours:

import google.cloud.logging
from datetime import datetime, timedelta, timezone
import os

# pull your project id from an environment variable
project_id = os.environ["GOOGLE_CLOUD_PROJECT"]
# construct a date object representing yesterday
yesterday = datetime.now(timezone.utc) - timedelta(days=1)
# Cloud Logging expects a timestamp in RFC3339 UTC "Zulu" format
# https://cloud.google.com/logging/docs/reference/v2/rest/v2/LogEntry
time_format = "%Y-%m-%dT%H:%M:%S.%f%z"
# build a filter that returns GKE Admin Activity audit Logs from
# the past 24 hours
# https://cloud.google.com/kubernetes-engine/docs/how-to/audit-logging
filter_str = (
    f'logName="projects/{project_id}/logs/cloudaudit.googleapis.com%2Factivity"'
    f' AND resource.type="k8s_cluster"'
    f' AND timestamp>="{yesterday.strftime(time_format)}"'
)
# query and print all matching logs
client = google.cloud.logging.Client()
for entry in client.list_entries(filter_=filter_str):
    print(entry)

Delete Log Entries

To delete all logs associated with a logger, use the following call:

logger.delete()  # API call

Manage Log Metrics

Logs-based metrics are counters of entries which match a given filter. They can be used within Cloud Monitoring to create charts and alerts.

To list all logs-based metrics for a project:

for metric in client.list_metrics():  # API call(s)
    do_something_with(metric)

To create a logs-based metric:

metric = client.metric(metric_name, filter_=filter, description=description)
assert not metric.exists()  # API call
metric.create()  # API call
assert metric.exists()  # API call

To refresh local information about a logs-based metric:

existing_metric = client.metric(metric_name)
existing_metric.reload()  # API call

To update a logs-based metric:

existing_metric.filter_ = updated_filter
existing_metric.description = updated_description
existing_metric.update()  # API call

To delete a logs-based metric:

    metric.delete()

Log Sinks

Sinks allow exporting of log entries which match a given filter to Cloud Storage buckets, BigQuery datasets, or Cloud Pub/Sub topics.

Cloud Storage Sink

Ensure the storage bucket that you want to export logs to has cloud-logs@google.com as an owner. See Setting permissions for Cloud Storage.

Ensure that cloud-logs@google.com is an owner of the bucket:

bucket.acl.reload()  # API call
logs_group = bucket.acl.group("cloud-logs@google.com")
logs_group.grant_owner()
bucket.acl.add_entity(logs_group)
bucket.acl.save()  # API call

To create a Cloud Storage sink:

destination = "storage.googleapis.com/%s" % (bucket.name,)
sink = client.sink(sink_name, filter_=filter, destination=destination)
assert not sink.exists()  # API call
sink.create()  # API call
assert sink.exists()  # API call

BigQuery Sink

To export logs to BigQuery, you must log into the Cloud Console and add cloud-logs@google.com to a dataset.

See: Setting permissions for BigQuery

from google.cloud.bigquery.dataset import AccessEntry

entry_list = dataset.access_entries
entry_list.append(AccessEntry("WRITER", "groupByEmail", "cloud-logs@google.com"))
dataset.access_entries = entry_list
client.update_dataset(dataset, ["access_entries"])  # API call

To create a BigQuery sink:

destination = "bigquery.googleapis.com%s" % (dataset.path,)
sink = client.sink(sink_name, filter_=filter_str, destination=destination)
assert not sink.exists()  # API call
sink.create()  # API call
assert sink.exists()  # API call

Pub/Sub Sink

To export logs to BigQuery you must log into the Cloud Console and add cloud-logs@google.com to a topic.

See: Setting permissions for Pub/Sub

topic_path = client.topic_path(project_id, topic_id)
topic = client.create_topic(request={"name": topic_path})

policy = client.get_iam_policy(request={"resource": topic_path})  # API call
policy.bindings.add(role="roles/owner", members=["group:cloud-logs@google.com"])

client.set_iam_policy(
    request={"resource": topic_path, "policy": policy}
)  # API call

To create a Cloud Pub/Sub sink:

destination = "pubsub.googleapis.com/%s" % (topic.name,)
sink = client.sink(sink_name, filter_=filter_str, destination=destination)
assert not sink.exists()  # API call
sink.create()  # API call
assert sink.exists()  # API call

Manage Sinks

To list all sinks for a project:

for sink in client.list_sinks():  # API call(s)
    do_something_with(sink)

To refresh local information about a sink:

existing_sink = client.sink(sink_name)
existing_sink.reload()

To update a sink:

existing_sink.filter_ = updated_filter
existing_sink.update()

To delete a sink:

sink.delete()