Aggiorna un cluster Google Cloud Managed Service per Apache Kafka

Puoi modificare un cluster Google Cloud Managed Service per Apache Kafka per aggiornare proprietà come le dimensioni del cluster, inclusi il numero di vCPU e la memoria, l'elenco delle subnet connesse, gli intervalli IP di origine consentiti per i cluster pubblici, la configurazione del ribilanciamento automatico e la configurazione mTLS.

Per modificare un cluster, puoi utilizzare la console Google Cloud , Google Cloud CLI, la libreria client o l'API Managed Kafka. Non puoi utilizzare l'API Apache Kafka open source per aggiornare un cluster.

L'aggiornamento di determinate proprietà, come il conteggio di vCPU e la memoria, potrebbe richiedere il riavvio del cluster da parte del servizio. Il servizio riavvia il cluster un broker alla volta. Durante questo processo, le richieste ai singoli broker potrebbero non andare a buon fine, ma questi errori sono temporanei. Le librerie client di uso comune gestiscono automaticamente questi errori.

Ruoli e autorizzazioni richiesti

Per ottenere le autorizzazioni necessarie per aggiornare un cluster, chiedi all'amministratore di concederti il ruolo IAM Editor cluster Kafka gestito (roles/managedkafka.clusterEditor) nel progetto. Per saperne di più sulla concessione dei ruoli, consulta Gestisci l'accesso a progetti, cartelle e organizzazioni.

Questo ruolo predefinito contiene le autorizzazioni necessarie per aggiornare un cluster. Per vedere quali sono esattamente le autorizzazioni richieste, espandi la sezione Autorizzazioni obbligatorie:

Autorizzazioni obbligatorie

Per aggiornare un cluster sono necessarie le seguenti autorizzazioni:

• Modifica un cluster: managedkafka.clusters.update

Potresti anche ottenere queste autorizzazioni con ruoli personalizzati o altri ruoli predefiniti.

Ridimensionare un cluster

Se aggiorni il numero di vCPU o la memoria di un cluster, si applicano le seguenti regole:

• Il rapporto complessivo tra vCPU e memoria del cluster deve sempre rimanere compreso tra 1:1 e 1:8.

• Deve essere presente almeno 1 vCPU e 1 GiB di memoria per ogni broker esistente. Il numero di broker non diminuisce mai.

• Se il cluster ha una configurazione del disco personalizzata, l'aggiornamento deve soddisfare i requisiti di configurazione del disco per l'archiviazione locale.

• Se esegui l'upgrade, la vCPU e la memoria medie per broker non possono diminuire di più del 10% rispetto alle medie prima dell'aggiornamento. Ad esempio, se provi a eseguire l'upscaling di un cluster da 45 vCPU (3 broker) a 48 vCPU (4 broker), la vCPU media per broker diminuisce da 15 a 12, con una riduzione del 20%, superiore al limite del 10%.

  Se devi ridurre il conteggio di vCPU di oltre il 10%, ti consigliamo di farlo in più fasi. Dopo ogni aggiornamento, monitora l'utilizzo delle risorse e ribilancia le partizioni, se necessario.

  Tuttavia, se ritieni che i tuoi broker avranno capacità sufficiente dopo l'aggiornamento, puoi disattivare questo controllo eseguendo il comando gcloud managed-kafka clusters update con il flag allow_broker_downscale_on_cluster_upscale=true. Questo flag indica che accetti il potenziale rischio per il rendimento.

Per ulteriori informazioni, vedi Aggiornare le dimensioni del cluster.

Configurazione del cluster pubblico

Puoi attivare o disattivare l'accesso pubblico per un cluster esistente, nonché aggiungere o rimuovere intervalli IP di origine consentiti. Per saperne di più sui requisiti e sulle regole per gli intervalli IP di origine consentiti, consulta Cluster pubblici.

Attenzione:per proteggere il cluster, consenti l'accesso solo da intervalli IP esterni attendibili. Non esporre il cluster a intervalli non attendibili.

Managed Service per Apache Kafka utilizza Cloud Next Generation Firewall per limitare l'accesso ai cluster pubblici. La rimozione degli intervalli IP di origine consentiti o la disattivazione dell'accesso pubblico si applica solo alle nuove connessioni. Per saperne di più, consulta Effetti sul traffico esistente.

Modifica di un cluster

Per modificare un cluster:

Console

1. Nella console Google Cloud , vai alla pagina Cluster.

Vai a Cluster

1. Nell'elenco dei cluster, fai clic su quello di cui vuoi modificare le proprietà.

La console mostra la pagina dei dettagli del cluster.

1. Nella pagina dei dettagli del cluster, fai clic su Modifica.

2. Modifica le proprietà in base alle esigenze. Puoi modificare le seguenti proprietà di un cluster dalla console:

   • Memoria
   • vCPU
   • Subnet
   • Configurazione del ribilanciamento
   • Configurazione mTLS
   • Etichette

3. Fai clic su Salva.

gcloud

1. Nella console Google Cloud , attiva Cloud Shell.

   Attiva Cloud Shell

   Nella parte inferiore della console Google Cloud viene avviata una sessione di Cloud Shell e viene visualizzato un prompt della riga di comando. Cloud Shell è un ambiente shell con Google Cloud CLI già installata e con valori già impostati per il progetto corrente. L'inizializzazione della sessione può richiedere alcuni secondi.

2. Prima di utilizzare i dati dei comandi riportati di seguito, effettua le seguenti sostituzioni:

   • PROJECT_ID: l'ID progetto
   • LOCATION: la posizione del cluster.
   • CLUSTER_ID: l'ID del cluster.
   • CPU_COUNT: il numero di vCPU per il cluster.
   • MEMORY: la quantità di memoria per il cluster. Esempio: 10GiB.
   • SUBNET_ID: l'ID subnet della subnet a cui connettersi. Esempio: default.
   • LABELS: le etichette da associare al cluster.
   • ALLOWED_SOURCE_IP_RANGES: gli intervalli CIDR IPv4 di origine consentiti per l'accesso a internet pubblico del cluster.

   Esegui questo comando:

   Linux, macOS o Cloud Shell

   Nota:assicurati di aver inizializzato Google Cloud CLI con l'autenticazione e un progetto eseguendo gcloud init; o gcloud auth login e gcloud config set project.

   gcloud managed-kafka clusters update CLUSTER_ID \
       --location=LOCATION \
       --cpu=CPU_COUNT \
       --memory=MEMORY \
       --subnets=projects/PROJECT_ID/regions/LOCATION/subnetworks/SUBNET_ID \
       --auto-rebalance \
       --labels=LABELS \
       --public-cluster \
       --allowed-source-ip-ranges=ALLOWED_SOURCE_IP_RANGES

   Windows (PowerShell)

   Nota:assicurati di aver inizializzato Google Cloud CLI con l'autenticazione e un progetto eseguendo gcloud init; o gcloud auth login e gcloud config set project.

   gcloud managed-kafka clusters update CLUSTER_ID `
       --location=LOCATION `
       --cpu=CPU_COUNT `
       --memory=MEMORY `
       --subnets=projects/PROJECT_ID/regions/LOCATION/subnetworks/SUBNET_ID `
       --auto-rebalance `
       --labels=LABELS `
       --public-cluster `
       --allowed-source-ip-ranges=ALLOWED_SOURCE_IP_RANGES

   Windows (cmd.exe)

   Nota:assicurati di aver inizializzato Google Cloud CLI con l'autenticazione e un progetto eseguendo gcloud init; o gcloud auth login e gcloud config set project.

   gcloud managed-kafka clusters update CLUSTER_ID ^
       --location=LOCATION ^
       --cpu=CPU_COUNT ^
       --memory=MEMORY ^
       --subnets=projects/PROJECT_ID/regions/LOCATION/subnetworks/SUBNET_ID ^
       --auto-rebalance ^
       --labels=LABELS ^
       --public-cluster ^
       --allowed-source-ip-ranges=ALLOWED_SOURCE_IP_RANGES

   Dovresti ricevere una risposta simile alla seguente:

   done: false
   metadata:
     '@type': type.googleapis.com/google.cloud.managedkafka.v1.OperationMetadata
     apiVersion: v1
     createTime: 'CREATE_TIME'
     requestedCancellation: false
     target: projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID
     verb: update
   name: projects/PROJECT_ID/locations/LOCATION/operations/OPERATION_ID
   

   • Per disattivare l'accesso pubblico, utilizza il flag --no-public-cluster.
   • Se utilizzi il flag --async con il comando, il sistema invia la richiesta di aggiornamento e restituisce immediatamente una risposta, senza attendere il completamento dell'operazione. Con il flag --async, puoi continuare con altre attività mentre l'aggiornamento del cluster viene eseguito in background. Se non utilizzi il flag --async, il sistema attende il completamento dell'operazione prima di restituire una risposta. Devi attendere il completamento dell'aggiornamento del cluster prima di poter continuare con altre attività.

REST

Prima di utilizzare i dati della richiesta, apporta le sostituzioni seguenti:

• PROJECT_ID: il tuo Google Cloud ID progetto
• LOCATION: la posizione del cluster.
• CLUSTER_ID: l'ID del cluster.
• UPDATE_MASK: quali campi aggiornare, come un elenco separato da virgole di nomi completi. Esempio: capacityConfig.vcpuCount,capacityConfig.memoryBytes
• CPU_COUNT: il numero di vCPU per il cluster.
• MEMORY: la quantità di memoria per il cluster, in byte. Esempio: 3221225472.
• SUBNET_ID: l'ID subnet della subnet a cui connettersi. Esempio: default.

Metodo HTTP e URL:

PATCH https://managedkafka.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID?updateMask=UPDATE_MASK

Corpo JSON della richiesta:

{
  "capacityConfig": {
    "vcpuCount": CPU_COUNT,
    "memoryBytes": MEMORY
  },
  "gcpConfig": {
    "accessConfig": {
      "networkConfigs": [
        {
          "subnet": "projects/PROJECT_ID/regions/LOCATION/subnetworks/SUBNET_ID"
        }
      ]
    }
  }
}

Per inviare la richiesta, espandi una di queste opzioni:

curl (Linux, macOS o Cloud Shell)

Nota: il seguente comando presuppone che tu abbia eseguito l'accesso all'interfaccia a riga di comando gcloud con il tuo account utente eseguendo gcloud init o gcloud auth login oppure utilizzando Cloud Shell, che consente di accedere automaticamente all'interfaccia a riga di comando gcloud. Puoi controllare l'account attualmente attivo eseguendo gcloud auth list.

Salva il corpo della richiesta in un file denominato request.json, quindi esegui il comando seguente:

curl -X PATCH \
     -H "Authorization: Bearer $(gcloud auth print-access-token)" \
     -H "Content-Type: application/json; charset=utf-8" \
     -d @request.json \
     "https://managedkafka.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID?updateMask=UPDATE_MASK"

PowerShell (Windows)

Nota: il comando seguente presuppone che tu abbia eseguito l'accesso all'interfaccia a riga di comando gcloud con il tuo account utente eseguendo gcloud init o gcloud auth login . Puoi controllare l'account attualmente attivo eseguendo gcloud auth list.

Salva il corpo della richiesta in un file denominato request.json, quindi esegui il comando seguente:

$cred = gcloud auth print-access-token
$headers = @{ "Authorization" = "Bearer $cred" }

Invoke-WebRequest `
    -Method PATCH `
    -Headers $headers `
    -ContentType: "application/json; charset=utf-8" `
    -InFile request.json `
    -Uri "https://managedkafka.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID?updateMask=UPDATE_MASK" | Select-Object -Expand Content

Dovresti ricevere una risposta JSON simile alla seguente:

{
  "name": "projects/PROJECT_ID/locations/LOCATION/operations/OPERATION_ID",
  "metadata": {
    "@type": "type.googleapis.com/google.cloud.managedkafka.v1.OperationMetadata",
    "createTime": "CREATE_TIME",
    "target": "projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID",
    "verb": "update",
    "requestedCancellation": false,
    "apiVersion": "v1"
  },
  "done": false
}

Nel corpo della richiesta, includi solo i campi che stai aggiornando, come specificato nel parametro di query UPDATE_MASK.

• Per aggiungere una subnet, aggiungi una nuova voce a networkConfigs nel seguente formato: projects/PROJECT_ID/regions/LOCATION/subnetworks/SUBNET_ID. Esempio: projects/sample-project/regions/us-central1/subnetworks/default.

• Per attivare l'accesso pubblico o aggiornare gli intervalli IP di origine consentiti, includi gcpConfig.accessConfig.publicClusterConfig nel parametro di query UPDATE_MASK e specifica l'array allowedSourceIpRanges nel corpo della richiesta. Esempio di corpo della richiesta:

  {
    "gcpConfig": {
      "accessConfig": {
        "publicClusterConfig": {
          "allowedSourceIpRanges": [
            "203.0.113.0/24"
          ]
        }
      }
    }
  }
  

• Per disattivare l'accesso pubblico, includi gcpConfig.accessConfig.publicClusterConfig nel parametro di query UPDATE_MASK e passa un oggetto JSON vuoto {} nel corpo della richiesta (o ometti publicClusterConfig). Corpo della richiesta di esempio:

  {}
  

Vai

Prima di provare questo esempio, segui le istruzioni di configurazione di Go in Installa le librerie client. Per saperne di più, consulta la documentazione di riferimento dell'API Go di Managed Service per Apache Kafka.

Per eseguire l'autenticazione in Managed Service per Apache Kafka, configura le Credenziali predefinite dell'applicazione(ADC). Per saperne di più, vedi Configura ADC per un ambiente di sviluppo locale.

import (
	"context"
	"fmt"
	"io"

	"cloud.google.com/go/managedkafka/apiv1/managedkafkapb"
	"google.golang.org/api/option"
	"google.golang.org/protobuf/types/known/fieldmaskpb"

	managedkafka "cloud.google.com/go/managedkafka/apiv1"
)

func updateCluster(w io.Writer, projectID, region, clusterID string, memory int64, opts ...option.ClientOption) error {
	// projectID := "my-project-id"
	// region := "us-central1"
	// clusterID := "my-cluster"
	// memoryBytes := 4221225472
	ctx := context.Background()
	client, err := managedkafka.NewClient(ctx, opts...)
	if err != nil {
		return fmt.Errorf("managedkafka.NewClient got err: %w", err)
	}
	defer client.Close()

	clusterPath := fmt.Sprintf("projects/%s/locations/%s/clusters/%s", projectID, region, clusterID)
	capacityConfig := &managedkafkapb.CapacityConfig{
		MemoryBytes: memory,
	}
	cluster := &managedkafkapb.Cluster{
		Name:           clusterPath,
		CapacityConfig: capacityConfig,
	}
	paths := []string{"capacity_config.memory_bytes"}
	updateMask := &fieldmaskpb.FieldMask{
		Paths: paths,
	}

	req := &managedkafkapb.UpdateClusterRequest{
		UpdateMask: updateMask,
		Cluster:    cluster,
	}
	op, err := client.UpdateCluster(ctx, req)
	if err != nil {
		return fmt.Errorf("client.UpdateCluster got err: %w", err)
	}
	resp, err := op.Wait(ctx)
	if err != nil {
		return fmt.Errorf("op.Wait got err: %w", err)
	}
	fmt.Fprintf(w, "Updated cluster: %#v\n", resp)
	return nil
}

Java

Prima di provare questo esempio, segui le istruzioni di configurazione di Java in Installare le librerie client. Per saperne di più, consulta la documentazione di riferimento dell'API Java di Managed Service per Apache Kafka.

Per eseguire l'autenticazione in Managed Service per Apache Kafka, configura le Credenziali predefinite dell'applicazione. Per saperne di più, consulta Configura ADC per un ambiente di sviluppo locale.

import com.google.api.gax.longrunning.OperationFuture;
import com.google.api.gax.longrunning.OperationSnapshot;
import com.google.api.gax.longrunning.OperationTimedPollAlgorithm;
import com.google.api.gax.retrying.RetrySettings;
import com.google.api.gax.retrying.TimedRetryAlgorithm;
import com.google.cloud.managedkafka.v1.CapacityConfig;
import com.google.cloud.managedkafka.v1.Cluster;
import com.google.cloud.managedkafka.v1.ClusterName;
import com.google.cloud.managedkafka.v1.ManagedKafkaClient;
import com.google.cloud.managedkafka.v1.ManagedKafkaSettings;
import com.google.cloud.managedkafka.v1.OperationMetadata;
import com.google.cloud.managedkafka.v1.UpdateClusterRequest;
import com.google.protobuf.FieldMask;
import java.time.Duration;
import java.util.concurrent.ExecutionException;

public class UpdateCluster {

  public static void main(String[] args) throws Exception {
    // TODO(developer): Replace these variables before running the example.
    String projectId = "my-project-id";
    String region = "my-region"; // e.g. us-east1
    String clusterId = "my-cluster";
    long memoryBytes = 25769803776L; // 24 GiB
    updateCluster(projectId, region, clusterId, memoryBytes);
  }

  public static void updateCluster(
      String projectId, String region, String clusterId, long memoryBytes) throws Exception {
    CapacityConfig capacityConfig = CapacityConfig.newBuilder().setMemoryBytes(memoryBytes).build();
    Cluster cluster =
        Cluster.newBuilder()
            .setName(ClusterName.of(projectId, region, clusterId).toString())
            .setCapacityConfig(capacityConfig)
            .build();
    FieldMask updateMask = FieldMask.newBuilder().addPaths("capacity_config.memory_bytes").build();

    // Create the settings to configure the timeout for polling operations
    ManagedKafkaSettings.Builder settingsBuilder = ManagedKafkaSettings.newBuilder();
    TimedRetryAlgorithm timedRetryAlgorithm = OperationTimedPollAlgorithm.create(
        RetrySettings.newBuilder()
            .setTotalTimeoutDuration(Duration.ofHours(1L))
            .build());
    settingsBuilder.updateClusterOperationSettings()
        .setPollingAlgorithm(timedRetryAlgorithm);

    try (ManagedKafkaClient managedKafkaClient = ManagedKafkaClient.create(
        settingsBuilder.build())) {
      UpdateClusterRequest request =
          UpdateClusterRequest.newBuilder().setUpdateMask(updateMask).setCluster(cluster).build();
      OperationFuture<Cluster, OperationMetadata> future =
          managedKafkaClient.updateClusterOperationCallable().futureCall(request);

      // Get the initial LRO and print details. CreateCluster contains sample code for polling logs.
      OperationSnapshot operation = future.getInitialFuture().get();
      System.out.printf("Cluster update started. Operation name: %s\nDone: %s\nMetadata: %s\n",
          operation.getName(),
          operation.isDone(),
          future.getMetadata().get().toString());

      Cluster response = future.get();
      System.out.printf("Updated cluster: %s\n", response.getName());
    } catch (ExecutionException e) {
      System.err.printf("managedKafkaClient.updateCluster got err: %s", e.getMessage());
    }
  }
}

Python

Prima di provare questo esempio, segui le istruzioni di configurazione di Python in Installare le librerie client. Per saperne di più, consulta la documentazione di riferimento dell'API Python di Managed Service per Apache Kafka.

Per eseguire l'autenticazione in Managed Service per Apache Kafka, configura le Credenziali predefinite dell'applicazione. Per saperne di più, vedi Configura ADC per un ambiente di sviluppo locale.

from google.api_core.exceptions import GoogleAPICallError
from google.cloud import managedkafka_v1
from google.protobuf import field_mask_pb2

# TODO(developer)
# project_id = "my-project-id"
# region = "us-central1"
# cluster_id = "my-cluster"
# memory_bytes = 4295000000

client = managedkafka_v1.ManagedKafkaClient()

cluster = managedkafka_v1.Cluster()
cluster.name = client.cluster_path(project_id, region, cluster_id)
cluster.capacity_config.memory_bytes = memory_bytes
update_mask = field_mask_pb2.FieldMask()
update_mask.paths.append("capacity_config.memory_bytes")

# For a list of editable fields, one can check https://cloud.google.com/managed-kafka/docs/create-cluster#properties.
request = managedkafka_v1.UpdateClusterRequest(
    update_mask=update_mask,
    cluster=cluster,
)

try:
    operation = client.update_cluster(request=request)
    print(f"Waiting for operation {operation.operation.name} to complete...")
    response = operation.result()
    print("Updated cluster:", response)
except GoogleAPICallError as e:
    print(f"The operation failed with error: {e.message}")

Limitazioni

Dopo aver creato un cluster Managed Service per Apache Kafka, non puoi aggiornare le seguenti proprietà:

• Nome del cluster
• La località del cluster
• Il tipo di crittografia

Anche se non puoi modificare il tipo di crittografia, puoi ruotare le chiavi di crittografia.

Passaggi successivi

• Monitora il tuo cluster.

• Pianifica le dimensioni del cluster.

• Crea un argomento all'interno del cluster.

• Visualizza il cluster.

• Elimina il cluster.

Apache Kafka® è un marchio registrato di Apache Software Foundation o dei suoi affiliati negli Stati Uniti e/o in altri paesi.