........
What is above is nothing more than a regular Flask web application. The Cloud Tasks invokes the two endpoints via POST. Both workers (generate the image and send the reply back to Discord) are deployed within the same webapp.
The only change is replacing "app_engine_http_request" with "http_request" and replacing "relative_uri" with "url". Exactly the same you can have workers deployed outside the Google Cloud and still be invoked via Cloud Tasks. That's one of the main differences between PUBSUB and Cloud Tasks. Via Cloud Tasks, it's the publisher which decides where the task is going to be executed. Via PUBSUB, the publisher has no knowledge where the task is going to be processed.
This post details the implementation of a highly scalable Discord bot using a single serverless function in GCP. It's part of a series of posts which describes the same project, a highly scalable Discord bot which reacts to the "/mando" command by generating a unique mandala and serving the image back, implemented using multiple approaches.
I did this to play and experiment with GCP. Might want to read the introductive post first. I've experimented with a single serverless function and it's not easy to implement what we want. Two serverless functions and a PUBSUB topic does the job well. Three serverless functions and two PUBSUB topics gives you even more freedom in allocating resources for each function, so that you can use a resource heavy instance for the function generating mandalas and very cheap instance for the other two. For this post, we are going to experiment with Cloud Tasks.
TLDR
- Cloud Tasks is a module which allows a service to asynchronously delegate work to other services. It's similar to PUBSUB, but the publisher has more control with Cloud Tasks over how the work is being completed. The publisher can specifically choose the consumer (for PUBSUB, the publisher is not aware of the consumers listening to a topic), scheduled delivery, rate control and message deduplication are possible
- As workers, you can use Cloud Functions, Cloud Run, GKE, Compute Engine, App Engine, or even an on-premise web server. The worker is nothing more but a regular webapp able to receive via POST, do the work, and return HTTP status codes to signal whether the task is done or has failed.
- In this post we're going to experiment with an App Engine deployment for the workers and also give a quick try to a serverless function as worker.
- When there's stable load on your system, it makes more sense to go for a GKE or App Engine deployment instead of serverless.
- You can deploy more than one worker in the same webapp, just make sure you have unique endpoints per each worker.
Cloud Task queues
gcloud tasks queues create create-image-queue
gcloud tasks queues create reply-discord-queueApp Engine workers
Deploying:
gcloud app deploySource code:
import uuid
from create_app_engine_queue_task import create_task
[...]app = Flask(__name__)
def upload_blob(source_file_name, destination_blob_name):
[....]
def post_cloud_task_send_reply(interaction_token, mando_filename):
mando_url = f"https://storage.googleapis.com/{bucket_name}/{mando_filename}"
discord_url = f"https://discord.com/api/v8/webhooks/{discord_app_id}/{interaction_token}/messages/@original"
data = {
"tts": False,
"content": "image",
"embeds": [
{
"thumbnail": {"url": mando_url},
"image": {"url": mando_url},
}
],
"allowed_mentions": {"parse": []},
}
payload = {
"discord_url": discord_url,
"data": data,
}
create_task(
project=project_name,
queue="reply-discord-queue-1",
location=region,
payload=payload,
relative_url="/reply-discord",
)
@app.route("/generate-image", methods=["POST"])
def generate_image():
payload_json = request.get_json()
local_file_name = f"{str(uuid.uuid4())}.png"
create_mando(local_tmp_folder=local_tmp_folder, local_file_name=local_file_name)
upload_blob(f"{local_tmp_folder}/{local_file_name}", local_file_name)
interaction_token = payload_json["interaction_token"]
post_cloud_task_send_reply(
interaction_token=interaction_token,
mando_filename=local_file_name,
)
return {}, 200
@app.route("/reply-discord", methods=["POST"])
def reply_discord():
payload_json = request.get_json()
discord_url = payload_json["discord_url"]
data = payload_json["data"]
reply_response = requests.patch(
url=discord_url,
json=data,
)
reply_response.raise_for_status()
return {}, 200
raw-discord-data function
The "raw-discord-data" serverless function is now a bit changed. Instead of posting data to PUBSUB, now it publish a task to the Cloud Tasks queue.
[...]
from create_app_engine_queue_task import create_task
def post_cloud_task_generate_image(interaction_id, interaction_token, message_id):
payload = {
"interaction_id": interaction_id,
"interaction_token": interaction_token,
"message_id": message_id,
}
create_task(
project=project_name,
queue="reply-discord-queue-1",
location=region,
payload=payload,
relative_url="/generate-image",
)
@functions_framework.http
def handle_request(request):
is_valid = validate_request(request)
if not is_valid:
abort(401, "invalid request signature")
request_type = request.json["type"]
if request_type == 1:
return jsonify({"type": 1})
else:
interaction_token = request.json["token"]
interaction_id = request.json["id"]
message_id = request.json["data"]["id"]
post_cloud_task_generate_image(interaction_id, interaction_token, message_id)
return jsonify(
{
"type": 4,
"data": {
"tts": False,
"content": "generating, please wait...",
"embeds": [],
"allowed_mentions": {"parse": []},
},
}
)
What if we replace the discord-reply AppEngine worker with a serverless function ?
Just for the sake of experimenting with GCP, let's replace the discord-reply worker (which send the PATCH request back to the Discord server when the image is ready) with a serverless function.
It's really simple. The only change I had to do was to update the payload published to the Cloud Tasks queue.
Payload to trigger the AppEngine worker:
task = {
"app_engine_http_request": {
"http_method": tasks_v2.HttpMethod.POST,
"relative_uri": app_engine_worker_relative_url,
}
}Payload to trigger the serverless function:
task = {
"http_request": {
"http_method": tasks_v2.HttpMethod.POST,
"url": serverless_function_absolute_url,
}
}
Comments
Post a Comment