Creazione gestori attività App Engine

Questa pagina illustra come creare un gestore delle attività di App Engine, il codice worker che gestisce un'attività di App Engine. La coda di Cloud Tasks invia richieste HTTP al gestore delle attività. Una volta completata l'elaborazione, il gestore deve inviare alla coda un codice di stato HTTP compreso tra 200 e 299. Qualsiasi altro valore indica che l'attività non è riuscita e la coda prova a ripetere l'attività.

Nota: App Engine restituisce un codice di stato 503 quando le istanze sono sovraccaricate o non disponibili per altri motivi. Il servizio Task risponde a questa richiesta rallentando la consegna dalle code ai gestori, per evitare di peggiorare il problema. Se vuoi attivare un nuovo tentativo intenzionalmente, utilizza un codice di stato diverso da 2xx o 503.

Le richieste della coda di attività di App Engine vengono inviate dall'indirizzo IP 0.1.0.2. Fai riferimento anche all'intervallo IP per le richieste inviate all'ambiente App Engine.

C#

cloudtasks/appengine/Startup.cs

Visualizza su GitHub Feedback

    public class Startup
    {
        public Startup(IConfiguration configuration)
        {
            Configuration = configuration;
        }

        public IConfiguration Configuration { get; }

        // This method gets called by the runtime. Use this method to add services to the container.
        // For more information on how to configure your application, visit https://go.microsoft.com/fwlink/?LinkID=398940
        public void ConfigureServices(IServiceCollection services)
        {
            services.AddLogging(builder => builder.AddDebug());
            services.AddRouting();
        }

        // This method gets called by the runtime. Use this method to configure the HTTP request pipeline.
        public void Configure(IApplicationBuilder app, IHostingEnvironment env, ILoggerFactory loggerFactory)
        {
            var logger = loggerFactory.CreateLogger("testStackdriverLogging");

            if (env.IsDevelopment())
            {
                app.UseDeveloperExceptionPage();
            }
            else
            {
                // Configure error reporting service.
                app.UseExceptionHandler("/Home/Error");
            }

            var routeBuilder = new RouteBuilder(app);

            routeBuilder.MapPost("log_payload", context =>
            {
                // Log the request payload
                var reader = new StreamReader(context.Request.Body);
                var task = reader.ReadToEnd();

                logger.LogInformation($"Received task with payload: {task}");
                return context.Response.WriteAsync($"Printed task payload: {task}");
            });

            routeBuilder.MapGet("hello", context =>
            {
                // Basic index to verify app is serving
                return context.Response.WriteAsync("Hello, world!");
            });

            routeBuilder.MapGet("_ah/health", context =>
            {
                // Respond to GAE health-checks
                return context.Response.WriteAsync("OK");
            });

            routeBuilder.MapGet("/", context =>
            {
                return context.Response.WriteAsync("Hello, world!");
            });

            var routes = routeBuilder.Build();
            app.UseRouter(routes);
        }
    }

Python

appengine/flexible_python37_and_earlier/tasks/main.py

Visualizza su GitHub Feedback

from flask import Flask, render_template, request

app = Flask(__name__)

@app.route("/example_task_handler", methods=["POST"])
def example_task_handler():
    """Log the request payload."""
    payload = request.get_data(as_text=True) or "(empty payload)"
    print(f"Received task with payload: {payload}")
    return render_template("index.html", payload=payload)

Java

appengine-java8/tasks/quickstart/src/main/java/com/example/task/TaskServlet.java

Visualizza su GitHub Feedback

@WebServlet(
    name = "Tasks",
    description = "Create Cloud Task",
    urlPatterns = "/tasks/create"
)
public class TaskServlet extends HttpServlet {
  private static Logger log = Logger.getLogger(TaskServlet.class.getName());

  @Override
  public void doPost(HttpServletRequest req, HttpServletResponse resp) throws IOException {
    log.info("Received task request: " + req.getServletPath());
    String body = req.getReader()
        .lines()
        .reduce("", (accumulator, actual) -> accumulator + actual);

    if (!body.isEmpty()) {
      log.info("Request payload: " + body);
      String output = String.format("Received task with payload %s", body);
      resp.getOutputStream().write(output.getBytes());
      log.info("Sending response: " + output);
      resp.setStatus(HttpServletResponse.SC_OK);
    } else {
      log.warning("Null payload received in request to " + req.getServletPath());
    }
  }
}

PHP

appengine/standard/tasks/apps/handler/index.php

Visualizza su GitHub Feedback

require __DIR__ . '/vendor/autoload.php';

use Google\Cloud\Logging\LoggingClient;

// Create the logging client.
$logging = new LoggingClient();
// Create a PSR-3-compatible logger.
$logger = $logging->psrLogger('app', ['batchEnabled' => true]);

// Front-controller to route requests.
switch (@parse_url($_SERVER['REQUEST_URI'])['path']) {
    case '/':
        print "Hello, World!\n";
        break;
    case '/task_handler':
        // Taskname and Queuename are two of several useful Cloud Tasks headers available on the request.
        $taskName = $_SERVER['HTTP_X_APPENGINE_TASKNAME'] ?? '';
        $queueName = $_SERVER['HTTP_X_APPENGINE_QUEUENAME'] ?? '';

        try {
            handle_task(
                $queueName,
                $taskName,
                file_get_contents('php://input')
            );
        } catch (Exception $e) {
            http_response_code(400);
            exit($e->getMessage());
        }
        break;
    default:
        http_response_code(404);
        exit('Not Found');
}

/**
 * Process a Cloud Tasks HTTP Request.
 *
 * @param string $queueName provides the name of the queue which dispatched the task.
 * @param string $taskName provides the identifier of the task.
 * @param string $body The task details from the HTTP request.
 */
function handle_task($queueName, $taskName, $body = '')
{
    global $logger;

    if (empty($taskName)) {
        // You may use the presence of the X-Appengine-Taskname header to validate
        // the request comes from Cloud Tasks.
        $logger->warning('Invalid Task: No X-Appengine-Taskname request header found');
        throw new Exception('Bad Request - Invalid Task');
    }

    $output = sprintf('Completed task: task queue(%s), task name(%s), payload(%s)', $queueName, $taskName, $body);
    $logger->info($output);

    // Set a non-2xx status code to indicate a failure in task processing that should be retried.
    // For example, http_response_code(500) to indicate a server error.
    print $output;
}

Go

appengine/go11x/tasks/handle_task/handle_task.go

Visualizza su GitHub Feedback

// Sample task_handler is an App Engine app demonstrating Cloud Tasks handling.
package main

import (
	"fmt"
	"io/ioutil"
	"log"
	"net/http"
	"os"
)

func main() {
	// Allow confirmation the task handling service is running.
	http.HandleFunc("/", indexHandler)

	// Handle all tasks.
	http.HandleFunc("/task_handler", taskHandler)

	port := os.Getenv("PORT")
	if port == "" {
		port = "8080"
		log.Printf("Defaulting to port %s", port)
	}

	log.Printf("Listening on port %s", port)
	if err := http.ListenAndServe(":"+port, nil); err != nil {
		log.Fatal(err)
	}
}

// indexHandler responds to requests with our greeting.
func indexHandler(w http.ResponseWriter, r *http.Request) {
	if r.URL.Path != "/" {
		http.NotFound(w, r)
		return
	}
	fmt.Fprint(w, "Hello, World!")
}

// taskHandler processes task requests.
func taskHandler(w http.ResponseWriter, r *http.Request) {
	taskName := r.Header.Get("X-Appengine-Taskname")
	if taskName == "" {
		// You may use the presence of the X-Appengine-Taskname header to validate
		// the request comes from Cloud Tasks.
		log.Println("Invalid Task: No X-Appengine-Taskname request header found")
		http.Error(w, "Bad Request - Invalid Task", http.StatusBadRequest)
		return
	}

	// Pull useful headers from Task request.
	queueName := r.Header.Get("X-Appengine-Queuename")

	// Extract the request body for further task details.
	body, err := ioutil.ReadAll(r.Body)
	if err != nil {
		log.Printf("ReadAll: %v", err)
		http.Error(w, "Internal Error", http.StatusInternalServerError)
		return
	}

	// Log & output details of the task.
	output := fmt.Sprintf("Completed task: task queue(%s), task name(%s), payload(%s)",
		queueName,
		taskName,
		string(body),
	)
	log.Println(output)

	// Set a non-2xx status code to indicate a failure in task processing that should be retried.
	// For example, http.Error(w, "Internal Server Error: Task Processing", http.StatusInternalServerError)
	fmt.Fprintln(w, output)
}

Node.js

cloud-tasks/snippets/server.js

Visualizza su GitHub Feedback

const express = require('express');

const app = express();
app.enable('trust proxy');

// Set the Content-Type of the Cloud Task to ensure compatibility
// By default, the Content-Type header of the Task request is set to "application/octet-stream"
// see https://cloud.google.com/tasks/docs/reference/rest/v2beta3/projects.locations.queues.tasks#AppEngineHttpRequest
app.use(express.text());

app.get('/', (req, res) => {
  // Basic index to verify app is serving
  res.send('Hello, World!').end();
});

app.post('/log_payload', (req, res) => {
  // Log the request payload
  console.log(`Received task with payload: ${req.body}`);
  res.send(`Printed task payload: ${req.body}`).end();
});

app.get('*', (req, res) => {
  res.send('OK').end();
});

const PORT = process.env.PORT || 8080;
app.listen(PORT, () => {
  console.log(`App listening on port ${PORT}`);
  console.log('Press Ctrl+C to quit.');
});

Ruby

appengine/cloud-tasks/app.rb

Visualizza su GitHub Feedback

require "sinatra"
require "json"

get "/" do
  # Basic index to verify app is serving
  "Hello World!"
end

post "/log_payload" do
  data = request.body.read
  # Log the request payload
  puts "Received task with payload: #{data}"
  "Printed task payload: #{data}"
end

Timeout

Le attività di App Engine hanno timeout specifici che dipendono dal tipo di scalabilità del servizio che le esegue.

Per i servizi worker in esecuzione nell'ambiente standard:

• Scalabilità automatica: l'elaborazione dell'attività deve terminare entro 10 minuti.
• Scalabilità manuale e di base: le richieste possono essere eseguite fino a 24 ore.

Per i servizi worker in esecuzione nell'ambiente flessibile: tutti i tipi hanno un timeout di 60 minuti.

Se il gestore non rispetta la scadenza, la coda presume che l'attività non sia riuscita e lo riprova.

Nota: valuta se il tuo gestore è idempotente. Cloud Tasks è progettato per fornire la consegna "almeno una volta"; ovvero, se un'attività viene aggiunta correttamente, la coda la consegna almeno una volta. Tieni presente che, in alcune rare circostanze, è possibile l'esecuzione di più attività, quindi il codice deve garantire che non ci siano effetti collaterali dannosi di un'esecuzione ripetuta.

Lettura delle intestazioni delle richieste delle attività di App Engine

Le richieste inviate al tuo gestore App Engine da una coda di Cloud Tasks hanno intestazioni speciali, che contengono informazioni specifiche per l'attività che il gestore potrebbe voler utilizzare.

Queste intestazioni sono impostate internamente. Se una di queste intestazioni è presente in una richiesta di un utente esterno alla tua app, viene sostituita da quelle interne, ad eccezione delle richieste provenienti dagli amministratori dell'applicazione che hanno eseguito l'accesso, che sono autorizzati a impostare intestazioni a scopo di test.

Le richieste delle attività di App Engine contengono sempre le seguenti intestazioni:

Titolo	Descrizione
X-AppEngine-QueueName	Il nome della coda.
X-AppEngine-TaskName	Il nome "breve" dell'attività o, se non è stato specificato alcun nome al momento della creazione, un ID univoco generato dal sistema. Questo è il valore my-task-id nel nome completo dell'attività, ad esempio task_name = projects/my-project-id/locations/my-location/queues/my-queue-id/tasks/my-task-id.
X-AppEngine-TaskRetryCount	Il numero di nuovi tentativi dell'attività. Per il primo tentativo, questo valore è 0. Questo numero include i tentativi in cui l'attività non è riuscita a causa della mancanza di istanze disponibili e non ha mai raggiunto la fase di esecuzione.
X-AppEngine-TaskExecutionCount	Il numero di volte in cui l'attività è stata eseguita e ha ricevuto una risposta dal gestore. Poiché Cloud Tasks elimina l'attività una volta ricevuta una risposta positiva, tutte le precedenti risposte del gestore sono errori. Questo numero non include gli errori dovuti alla mancanza di istanze disponibili. Tieni presente che X-AppEngine-TaskExecutionCount può essere uguale a X-AppEngine-TaskRetryCount se viene aggiornato prima del tentativo di esecuzione.
X-AppEngine-TaskETA	L'ora programmata dell'attività, specificata in secondi dal 1° gennaio 1970.

Se il gestore delle richieste trova una delle intestazioni elencate sopra, può considerare attendibile che la richiesta sia una richiesta Cloud Tasks.

Inoltre, le richieste da Cloud Tasks potrebbero contenere le seguenti intestazioni:

Titolo	Descrizione
X-AppEngine-TaskPreviousResponse	Il codice di risposta HTTP del tentativo precedente.
X-AppEngine-TaskRetryReason	Il motivo per cui hai riprovato a eseguire l'attività.
X-AppEngine-FailFast	Indica che un'attività non riesce immediatamente se non è disponibile un'istanza esistente.

Routing di destinazione

Nelle attività di App Engine, sia la coda sia il gestore delle attività vengono eseguiti all'interno dello stesso progetto Google Cloud. Il traffico è criptato durante il trasporto e non lascia mai i data center di Google. Non puoi impostare esplicitamente il protocollo (ad esempio HTTP o HTTPS). Tuttavia, la richiesta al gestore sembrerà che abbia utilizzato il protocollo HTTP.

Le attività possono essere inviate a gestori di attività sicuri, gestori di attività non sicuri e, nei runtime supportati, URI limitati con login: admin. Poiché le attività non vengono eseguite come utenti, non possono essere inviate a URI limitati con login: required. Inoltre, gli invii di attività non seguono i reindirizzamenti.

Passaggi successivi

• Scopri di più sulle attività nel riferimento per le API RPC.
• Scopri di più sulle attività nel riferimento sull'API REST.