Forem: Kenji Koshikawa

Manage private Python packages using Artifact Registry (Google Cloud)

Kenji Koshikawa — Sun, 06 Mar 2022 05:30:06 +0000

Lately, I build a system over microservices on Google Cloud. I often use Python with App Engine, Cloud Functions and Cloud Run.

There are some private libraries in common across each microservice.
Referring to here official documents(App Engine, Cloud Functions), we had deployed each microservice using the pip command pip install -t to copy dependencies into a local folder.

After a couple of week since introducing the above pip command operation when deploying, I realized that it was hard to control a system quality, because the above pip command was difficult to specify a dependency version each library in our team environments. So we could not feel free to update some libraries which had been used by many microservices.

In this article, I introduce how manage private Python packages using Artifact Registry and solve those quality control issues.

All source code which I wrote about in this article has been uploaded below for your reference.
https://github.com/koshilife/artifact-registry-python-repo-example

Set up tools and a virtual environment

The following are versions of gcloud, Python and direnv when I tried.

$ gcloud app update
$ gcloud --version
Google Cloud SDK 374.0.0
alpha 2022.02.22
beta 2022.02.22
core 2022.02.22
gsutil 5.6
kubectl 1.21.9

$ python3 --version
Python 3.9.10

$ direnv --version
2.28.0

I made a directory for an example app and set up a Python virtual environment. You need twine which is for publishing Python and keyrings.google-artifactregistry-auth which handles authentication with Artifact Registry repositories. The official guide is here for your reference.

$ mkdir hello-artifact-registry
$ cd hello-artifact-registry

$ python3 -m venv venv
$ source venv/bin/activate
$ python3 -m pip install --upgrade pip
$ pip install --upgrade pip setuptools wheel
$ pip install twine keyrings.google-artifactregistry-auth

The following is a direnv file.

export GCP_PROJECT="<YOUR-GCP-PROJECT>"

# a credential file path of service account which is granted read/write for Artifact Registry
export GOOGLE_APPLICATION_CREDENTIALS="<YOUR-PATH>/google_cloud_credntials.json"

Create a repository on Artifact Registry

I made a repository artifact-example for Python library at Tokyo region using the below gcloud command. The related official guide is here

$ gcloud artifacts repositories create artifact-example \
    --repository-format=python \
    --location=asia-northeast1 \
    --description="Python private libraries"

Upload some packages to the repository

For testing I prepared very simple python libraries which just return foo or bar string.

A directory structure of these libraries is below.

python_libs
├── artifact-example-foo
│   ├── artifact_example_foo
│   │   └── __init__.py
│   ├── requirements.txt
│   └── setup.py
└── artifact-example-bar
    ├── artifact_example_bar
    │   └── __init__.py
    ├── requirements.txt
    └── setup.py

artifact-example-foo (v1)
init.py

__version__ = '0.0.1'

def foo():
    return 'foo'

artifact-example-bar (v1)
init.py

__version__ = '0.0.1'

def bar():
    return 'bar'

I packaged as wheel format and uploaded it to the repository as the below commands.

$ cd python_libs/artifact-example-foo
$ python setup.py bdist_wheel
$ twine upload --repository-url https://asia-northeast1-python.pkg.dev/${GCP_PROJECT}/artifact-example/ dist/*

$ cd python_libs/artifact-example-bar
$ python setup.py bdist_wheel
$ twine upload --repository-url https://asia-northeast1-python.pkg.dev/${GCP_PROJECT}/artifact-example/ dist/*

I made additional versions each library for testing as v2. On the same way I packaged and uploaded to the repository.

artifact-example-foo (v2)

__version__ = '0.0.2'

def foo():
    return 'foo'

def foo2():
    return 'foo2'

artifact-example-bar (v2)

__version__ = '0.0.2'

def bar():
    return 'bar'

def bar2():
    return 'bar2'

Use these uploaded libraries in client Python apps

I created simple Python apps (Flask) in App Engine, Cloud Functions and Cloud Run in the same Google Cloud project as the repository of Artifact Registry.

Notice: If you create in another Google Cloud project, you might need to grant a service account of Code Build. (The related official guide is here)

App Engine

client_apps/appengine/main.py

from flask import Flask, jsonify
import artifact_example_foo as foo
import artifact_example_bar as bar

app = Flask(__name__)

@app.route('/')
def hello():
    return jsonify(dict(result='ok', foo=foo.foo(), foo_version=foo.__version__, bar=bar.bar2(), bar_version=bar.__version__))

if __name__ == '__main__':
    app.run(host='127.0.0.1', port=8080, debug=True)

client_apps/appengine/requirements.txt

--extra-index-url https://asia-northeast1-python.pkg.dev/<YOUR-GCP-PROJECT>/artifact-example/simple
artifact-example-foo==0.0.1
artifact-example-bar
Flask==2.0.2

In the requirements.txt file, a point is that the Artifact Registry URL is added into the first line and the relevant packages to include private dependencies artifact-example-foo and artifact-example-bar.

Also, the /simple string at the end of the repository URL is needed because of Python Simple Repository API.

In the dependencies body, for testing I specified artifact-example-foo to 0.0.1 (v1) and do not specify any versions in artifact-example-bar

https://cloud.google.com/appengine/docs/standard/python3/specifying-dependencies#private_dependencies_with_artifact_registry

Cloud Functions

client_apps/functions/main.py

from flask import jsonify

import artifact_example_foo as foo
import artifact_example_bar as bar

def main(request):
    return jsonify(dict(result='ok', foo=foo.foo(), foo_version=foo.__version__, bar=bar.bar2(), bar_version=bar.__version__))

client_apps/functions/requirements.txt

--extra-index-url https://asia-northeast1-python.pkg.dev/<YOUR-GCP-PROJECT>/artifact-example/simple
artifact-example-foo==0.0.1
artifact-example-bar

The method of declaring the Artifact Registry repository is the same as the above App Engine one.

https://cloud.google.com/functions/docs/writing/specifying-dependencies-python#private_dependencies_from_artifact_registry

Cloud Run

client_apps/cloudrun/main.py

import os

from flask import Flask, jsonify
import artifact_example_foo as foo
import artifact_example_bar as bar

app = Flask(__name__)

@app.route("/")
def main():
    return jsonify(dict(result='ok', foo=foo.foo(), foo_version=foo.__version__, bar=bar.bar2(), bar_version=bar.__version__))

if __name__ == "__main__":
    app.run(debug=True, host="0.0.0.0", port=int(os.environ.get("PORT", 8080)))

client_apps/cloudrun/requirements.txt

--extra-index-url https://asia-northeast1-python.pkg.dev/<YOUR-GCP-PROJECT>/artifact-example/simple
artifact-example-foo==0.0.1
artifact-example-bar
Flask==2.0.2
gunicorn==20.1.0

The method of declaring the Artifact Registry repository is the same as the above too.

client_apps/cloudrun/Dockerfile

# refs: https://cloud.google.com/run/docs/quickstarts/build-and-deploy/deploy-python-service

FROM python:3.10-slim

ENV PYTHONUNBUFFERED True
ENV GOOGLE_APPLICATION_CREDENTIALS "key.json"

ENV APP_HOME /app
WORKDIR $APP_HOME
COPY . ./

RUN pip install --upgrade pip
RUN pip install keyrings.google-artifactregistry-auth
RUN pip install --no-cache-dir -r requirements.txt

CMD exec gunicorn --bind :$PORT --workers 1 --threads 8 --timeout 0 main:app

I needed to enables to access the Artifact Registry repository while installing dependencies in Docker, so the following command allowed to copy the service account credentials.

ENV GOOGLE_APPLICATION_CREDENTIALS "key.json"

The deploy script is for your reference.

The Response of these client apps

The following JSON is the response of theses client apps.
You can see that the library foo's version is used 0.0.1 (v1) and the library bar's version is used 0.0.2 (v2) as expected.

{
    "bar": "bar2",
    "bar_version": "0.0.2",
    "foo": "foo",
    "foo_version": "0.0.1",
    "result": "ok"
}

Conclusion

By using Artifact Registry I could specify a version each library, so I would improve to manage python packages and to control a system quality finely than current the pip command pip install -t.

Other references

Associate with parent Cloud Workflows logs and child APIs logs using structured logs

Kenji Koshikawa — Fri, 04 Mar 2022 11:28:17 +0000

Lately, I build a system using Cloud Workflows which can combine Google Cloud Services such as Cloud Functions and Cloud Run. Sometimes, I was in a situation where I want to examine more efficiently using logs on Cloud Logging when debugging or daily monitoring.

In a workflow, You can easily save logs to Cloud Logging using sys.log. Also, you can see API logs on Cloud Logging which be called by the inside of Workflow. But these logs don't associate with each other, so it is hard to trace across parent workflow logs and child API logs from top to bottom orderly. It was tough when I met in high traffic cases. I associated manually using each log's timestamp as a clue.

In this article, I introduce to associate parent workflow logs and child API logs using a workflow execution id and structured logs.

An example app

I created a simple workflow which calls Cloud Functions and Cloud Run each one time. These APIs used Flask.

All source code which I wrote about in this article has been uploaded below for your reference.
https://github.com/koshilife/cloud-workflows-logging-example

Cloud Workflows

The following code is the workflow.

workflows/v1.yml

main:
    params: [input]
    steps:
    - set_trace_id:
        assign:
            - execution_id: ${sys.get_env("GOOGLE_CLOUD_WORKFLOW_EXECUTION_ID")}
            - api_url_function: "https://asia-northeast1-xxx.cloudfunctions.net/logging-example-v1" # TODO set your environment
            - api_url_cloudrun: "https://logging-example-v1-xxx-an.a.run.app" # TODO set your environment
    - call_function:
        call: http.post
        args:
            url: ${api_url_function}
            body:
                workflow_execution_id: ${execution_id}
            auth:
                type: OIDC
        result: result_function
    - call_cloudrun:
        call: http.post
        args:
            url: ${api_url_cloudrun}
            body:
                workflow_execution_id: ${execution_id}
            auth:
                type: OIDC
        result: result_cloudrun
    - set_result:
        assign:
            - result_map:
                message: ${execution_id}
                function_execution_id: ${result_function.headers["Function-Execution-Id"]}
                function_trace_id: ${text.split(result_function.headers["X-Cloud-Trace-Context"], ";o=")[0]}
                cloudrun_trace_id: ${text.split(result_cloudrun.headers["X-Cloud-Trace-Context"], ";o=")[0]}
    - log_result:
        call: sys.log
        args:
            json: ${result_map}
    - return_output:
        return: ${result_map}

${sys.get_env("GOOGLE_CLOUD_WORKFLOW_EXECUTION_ID")} gets a workflow execution ID and then the ID is set into each API request params.

In addition, I included each API Trace ID and Functions execution ID that hope to be a clue for examining.

Cloud Functions

The following code is the Cloud Functions.

functions/v1/main.py

import json

from flask import jsonify, request

GCP_PROJECT = os.environ.get('GCP_PROJECT')

def main(request):
    log('Cloud Functions msg1')
    log('Cloud Functions msg1', 'WARNING')
    return jsonify(dict(result='ok'))


def log(message, severity='INFO'):
    log_fields = {}

    # refs: https://cloud.google.com/functions/docs/monitoring/logging#writing_structured_logs
    if request and request.headers.get("X-Cloud-Trace-Context") and GCP_PROJECT:
        trace = request.headers["X-Cloud-Trace-Context"].split("/")
        log_fields["logging.googleapis.com/trace"] = f"projects/{GCP_PROJECT}/traces/{trace[0]}"

    request_json = request.get_json()
    if request_json and request_json.get("workflow_execution_id"):
        log_fields["logging.googleapis.com/labels"] = {'workflows.googleapis.com/execution_id': request_json["workflow_execution_id"]}

    payload = dict(message=message, severity=severity, **log_fields)
    print(json.dumps(payload))

The key point is using structured logs.
I set a field logging.googleapis.com/labels in each log entry to the workflow execution ID from the parent workflow.

refs: Special fields in structured payloads

Also, a reason why to specify workflows.googleapis.com/execution_id as a key of the labels field is that workflow execution logs use the same key.

Cloud Run

The following code is the Cloud Run.

cloudruns/v1/main.py

import os
import json

from flask import Flask, jsonify, request

GCP_PROJECT = os.environ.get('GCP_PROJECT')

app = Flask(__name__)


@app.route("/", methods=["POST"])
def main():
    log('Cloud Run msg1')
    log('Cloud Run msg1', 'WARNING')
    return jsonify(dict(result='ok'))


def log(message, severity='INFO'):
    log_fields = {}

    # refs: https://cloud.google.com/functions/docs/monitoring/logging#writing_structured_logs
    if request and request.headers.get("X-Cloud-Trace-Context") and GCP_PROJECT:
        trace = request.headers["X-Cloud-Trace-Context"].split("/")
        log_fields["logging.googleapis.com/trace"] = f"projects/{GCP_PROJECT}/traces/{trace[0]}"

    request_json = request.get_json()
    if request_json and request_json.get("workflow_execution_id"):
        log_fields["logging.googleapis.com/labels"] = {'workflows.googleapis.com/execution_id': request_json["workflow_execution_id"]}

    payload = dict(message=message, severity=severity, **log_fields)
    print(json.dumps(payload))


if __name__ == "__main__":
    app.run(debug=True, host="0.0.0.0", port=int(os.environ.get("PORT", 8080)))

The point is the same as mentioned above at Cloud Functions.
I used structured logs and put the parent workflow execution ID.

Result

After executing the workflow, I could retrieve across the parent workflow logs and child APIs logs by the following query on Cloud Logging Console.

labels."workflows.googleapis.com/execution_id"="<Workflow Execution ID>" OR
labels.execution_id="<Workflow Execution ID>"

I tried an another example app which is more logs than above. (refs: The app name is v2. The source code is here)

I could see that all logs as below.

(Addition)

I wanted to simplify the query on Cloud Logging like one-liner labels."workflows.googleapis.com/execution_id"="<Workflow Execution ID>", so I tried to change sys.log params to use structured logs in the workflow. But It didn't work. I guess Cloud Workflows may not support structured logs yet.

Following code is sub-workflow which tries to use structured logs. (all source code is here)

log:
    params: [message]
    steps:
    - set_log_data:
        assign:
            - execution_id_map: {}
            - execution_id_map["workflows.googleapis.com/execution_id"]: ${sys.get_env("GOOGLE_CLOUD_WORKFLOW_EXECUTION_ID")}
            - data: {}
            - data["message"]: ${message}
            - data["logging.googleapis.com/labels"]: ${execution_id_map}
    - call_sys_log:
        call: sys.log
        args:
            json: ${data}

LogEntry

Retry automatically with Exponential Backoff in Cloud Workflows

Kenji Koshikawa — Sun, 13 Feb 2022 14:39:55 +0000

Recently I build workflows using Cloud Workflows to combine some modules of Cloud Functions or Cloud Run on Google Cloud.

Sometimes I encountered errors of 429 or 500 related scaling issues when calling an endpoint of Cloud Functions or Cloud Run.

According to this official document, the solution of retrying with exponential backoff is introduced.

The solution:
For HTTP trigger-based functions, have the client implement exponential backoff and retries for requests that must not be dropped.

So this post introduces the solution in Cloud Workflows.

An example of Cloud Functions

I made a simple function of Cloud Functions to reproduce the scaling errors easily like below.

just sleeps 3 seconds
set scale settings to the minimum　(min instance:0, max instance:1)

foobar/main.py

import time

import flask

def main(request):
    time.sleep(3)
    return flask.jsonify({'result': 'ok'})

Following command can deploy to Cloud Functions.

# Deploys the function
$ gcloud functions deploy foobar \
  --entry-point main \
  --runtime python39 \
  --trigger-http \
  --region asia-northeast1 \
  --timeout 120 \
  --memory 128MB \
  --min-instances 0 \
  --max-instances 1 \
  --source ./foobar

# Grants a service account associated with workflows to execute the function
$ gcloud functions add-iam-policy-binding foobar \
    --region=asia-northeast1 \
    --member=serviceAccount:${YOUR-SERVICE-ACCOUNT} \
    --role=roles/cloudfunctions.invoker

A workflow for reproducing the scaling errors

At first, the following code is a workflow for reproducing the scaling errors.

main:
    params: [input]
    steps:
    - callFunc:
        call: http.get
        args:
            url: https://asia-northeast1-xxx.cloudfunctions.net/foobar
            auth:
                type: OIDC
        result: api_result
    - returnOutput:
        return: ${api_result.body}

Following command can deploy to Cloud Workflows.

$ gcloud workflows deploy v1 \
                --source=v1.yml \
                --location=asia-southeast1 \
                --service-account=${YOUR-SERVICE-ACCOUNT}

In order to reproduce the scaling errors, I executed the following shell over 20 times.

$ gcloud workflows run --project=${YOUR-PROJECT} --location=asia-southeast1 v1 --data='{}' &

As expected, 429 error was reproduced many times. The probability of success of the workflow executions was about 6 in 20.

In the console of Cloud Workflows and Cloud Functions, I could see below an error Information.

HTTP server responded with error code 429
in step "callFunc", routine "main", line: 5
{
  "body": "Rate exceeded.",
  "code": 429,
  "headers": {
    "Alt-Svc": "h3=\":443\"; ma=2592000,h3-29=\":443\"; ma=2592000,h3-Q050=\":443\"; ma=2592000,h3-Q046=\":443\"; ma=2592000,h3-Q043=\":443\"; ma=2592000,quic=\":443\"; ma=2592000; v=\"46,43\"",
    "Content-Length": "14",
    "Content-Type": "text/html",
    "Date": "Wed, 09 Feb 2022 08:17:19 GMT",
    "Server": "Google Frontend",
    "X-Cloud-Trace-Context": "2a8e4ba95570e4a6585a0b678d7f3b98"
  },
  "message": "HTTP server responded with error code 429",
  "tags": [
    "HttpError"
  ]
}

The solution

Next, The following code is the solution to retry automatically with exponential backoff.

In sub-workflow call_api, it is implemented to retry with exponential backoff when returning HTTP status 429 or 500 code. I set the retry count to 5 times and the initial sleep time to 10 seconds.

main:
    params: [input]
    steps:
    - callFunc:
        call: call_api
        args:
            url: https://asia-northeast1-xxx.cloudfunctions.net/foobar
        result: api_result
    - return_output:
        return: ${api_result.body}

call_api:
    params: [url]
    steps:
        - setup:
            assign:
                - retry_count: 5
                - first_sleep_sec: 10
                - sleep_time: ${first_sleep_sec}
        - try_many_times:
            for:
                value: count
                range: [1, ${retry_count}]
                steps:
                    - log_before_call:
                        call: sys.log
                        args:
                            text: ${"call_api url=" + url + " (" + string(count) + "/" + string(retry_count) + ")"}
                    - try_call_block:
                        try:
                            steps:
                                - request_url:
                                    call: http.get
                                    args:
                                        url: ${url}
                                        auth:
                                            type: OIDC
                                    result: api_result
                                - return_result:
                                    return: ${api_result}
                        except:
                            as: e
                            steps:
                                - handle_error:
                                    switch:
                                        - condition: ${count >= retry_count}
                                          raise: ${e}
                                        - condition: ${not("HttpError" in e.tags)}
                                          raise: ${e}
                                        - condition: ${(e.code == 429 or e.code == 500)}
                                          next: log_sleep_time
                                        - condition: true
                                          raise: ${e}
                                - log_sleep_time:
                                    call: sys.log
                                    args:
                                        severity: 'WARNING'
                                        text: ${"got HTTP status " + string(e.code) + ". waiting " + string(sleep_time) + " seconds."}
                                - wait:
                                    call: sys.sleep
                                    args:
                                        seconds: ${sleep_time}
                                - update_sleep_time:
                                    assign:
                                        - sleep_time: ${sleep_time * 2}
                                - next_continue:
                                    next: continue

In the same way, after deploying this workflow I executed it over 20 times.

As a result, All of the workflow executions succeeded and improved. One of them took 3 minutes over, but the retries worked expected.

According to the logs, I could see to wait exponentially each retry time like 10, 20, 40 and 80 seconds so on.

Conclusion

This post introduced an implementation to retry automatically with exponential backoff in Cloud Workflows and showed the solution is effective and enables to continue processing when occurring scaling problems.

(Addition) A simpler way using try/retry statement

There is a simpler way to use try/retry statement. Thanks to @krisbraun who shared at this article's comment.

Pattern1: using default retry policy (very simple)

If your function is idempotent, I think most of the use cases can be covered by the default ${http.default_retry}.

Simple default retry policy for idempotent targets.
Retries on 429 (Too Many Requests), 502 (Bad Gateway), 503 (Service unavailable), and 504 (Gateway Timeout), as well as on any ConnectionError and TimeoutError.
Uses max_retries of 5, and backoff as per retry.default_backoff.

main:
    params: [input]
    steps:
    - call_api:
        try:
            call: http.get
            args:
                url: https://asia-northeast1-xxx.cloudfunctions.net/foobar
                auth:
                    type: OIDC
            result: api_result
        retry: ${http.default_retry}
    - return_value:
        return: ${api_result.body}

Notice: ${http.default_retry} doesn't retry when 500 code.

Pattern2: using custom policy

The following workflow shows to add a retry condition when occurring 500 code in addition to ${http.default_retry} retry conditions. Also, the custom setting for backoff is set to initial sleep time 10 seconds, multiplier 2.

main:
    params: [input]
    steps:
    - call_api:
        try:
            call: http.get
            args:
                url: https://asia-northeast1-xxx.cloudfunctions.net/foobar
                auth:
                    type: OIDC
            result: api_result
        retry:
            predicate: ${custom_retry_policy}
            backoff:
                initial_delay: 10
                max_delay: 300
                multiplier: 2
    - return_value:
        return: ${api_result.body}

custom_retry_policy:
    params: [e]
    steps:
    - assign_retry_codes:
        assign:
            - retry_codes: [429, 500, 502, 503, 504]
    - what_to_repeat:
        switch:
          - condition: ${("code" in e) and (e.code in retry_codes)}
            return: True
          - condition: ${("tags" in e) and ("ConnectionError" in e.tags)}
            return: True
          - condition: ${("tags" in e) and ("TimeoutError" in e.tags)}
            return: True
    - otherwise:
          return: False

Which is a better way of saving Cloud Logging for Python in Cloud Functions?

Kenji Koshikawa — Sat, 12 Feb 2022 11:09:04 +0000

There were multiple ways to save logs from Cloud Functions to Cloud Logging using Python, so I tried to organize the merits and demerits in a simple way.

TL;DR

ways	1.Standard Logger	2.Structured logging	3.Official logging handlers	4.Using Cloud Logging API directly
no needs to extra libraries	✔	✔	-	-
allows log severity	-	✔	✔	✔
allows to add other fields	-	✔	-	✔
allows Python 3.7	✔	-	✔	✔

1. using Standard Logging

This is the most simple way. You can use the standard logger, but it doesn’t support the severity.

The following code is an example using the standard logger in Cloud Functions.

import logging

import flask

logger = logging.getLogger()
logger.setLevel(logging.DEBUG)

def main(request):
    print("PRINT message")
    logger.debug("DEBUG message")
    logger.info("INFO message")
    logger.warning("WARNING message")
    logger.error("ERROR message")
    logger.critical("CRITICAL message")
    logger.exception("EXCEPTION message")
    return flask.jsonify({'result': 'ok'})

The following was the actual screenshot from the Cloud Functions console.
You can see that all severity of entries are default.

2. using Structured logging

If you want to include log levels or other specific fields in your log entries, you can write logs to stdout or stderr in the form of a single line of serialized JSON. The Structured logging support is available in Python 3.8 and later.

https://cloud.google.com/functions/docs/monitoring/logging#writing_structured_logs

The following code is an example using the Structured logging in Cloud Functions.

import logging

import flask

logger = logging.getLogger()
logger.setLevel(logging.DEBUG)

def main(request):
    print("PRINT message")
    logger.debug("DEBUG message")
    logger.info("INFO message")
    logger.warning("WARNING message")
    logger.error("ERROR message")
    logger.critical("CRITICAL message")
    logger.exception("EXCEPTION message")
    return flask.jsonify({'result': 'ok'})

The following was the actual screenshot from the Cloud Functions console.
You can see that all severity has been correctly reflected.

When I tried on Python 3.7 runtime, the severity didn’t support below as described in Official Doc.

3. using Official logging handlers

Logging handlers with Python loggers are provided by the official library.

https://cloud.google.com/logging/docs/setup/python#configuring_the_logging_handler

The following code is an example using the logging handlers in Cloud Functions. (The library version of google-cloud-logging I used was 2.7.0. )

import logging

import flask
import google.cloud.logging

client = google.cloud.logging.Client()
client.setup_logging()
logger = logging.getLogger()
logger.setLevel(logging.DEBUG)

def main(request):
    print("PRINT message")
    logger.debug("DEBUG message")
    logger.info("INFO message")
    logger.warning("WARNING message")
    logger.error("ERROR message")
    logger.critical("CRITICAL message")
    logger.exception("EXCEPTION message")
    return flask.jsonify({'result': 'ok'})

The following was the actual screenshot from the Cloud Functions console.
You can see that all severity has been correctly reflected.

4. using Cloud Logging API directly

The official Doc also explains how to run the API directly using the Cloud Logging API client.

https://cloud.google.com/logging/docs/setup/python#use_the_cloud_client_library_directly

The following code is an example using the calling directory in Cloud Functions. (The library version of google-cloud-logging I used was 2.7.0. )

import flask
from google.cloud import logging

logging_client = logging.Client()
log_name = "hello-logging-func4"
logger = logging_client.logger(log_name)

def main(request):
    print("PRINT message")
    logger.log_text("DEFAULT message", severity="DEFAULT")
    logger.log_text("DEBUG message", severity="DEBUG")
    logger.log_text("INFO message", severity="INFO")
    logger.log_text("NOTICE message", severity="NOTICE")
    logger.log_text("WARNING message", severity="WARNING")
    logger.log_text("ERROR message", severity="ERROR")
    logger.log_text("CRITICAL message", severity="CRITICAL")
    logger.log_text("ALERT message", severity="ALERT")
    logger.log_text("EMERGENCY message", severity="EMERGENCY")
    return flask.jsonify({'result': 'ok'})

The following was the actual screenshot from the Cloud Functions console, but you cannot see all of the log entries.

You can see all of the log entries from the Cloud Logging console.

Conclusion

This post introduced ways of saving Cloud Logging for Python in Cloud Functions and organized the merits and demerits in a simple way.

If you don't need a log level, I recommend using Python Standard Logger(No.1).

If you need a log level, I recommend using official logging handlers(No.3) because you can setup easily. Or I recommend using Structured logging(No.2) because it is very flexible.