Crea un clúster de Google Cloud Managed Service para Apache Kafka

Un clúster de Managed Service para Apache Kafka proporciona un entorno para almacenar y procesar flujos de mensajes organizados en temas.

Para crear un clúster, puedes usar la consola de Google Cloud , Google Cloud CLI, la biblioteca cliente o la API de Kafka administrado. No puedes usar la API de Apache Kafka de código abierto para crear un clúster.

Antes de comenzar

Verifica que conozcas los siguientes conceptos:

• Cómo funciona Managed Service para Apache Kafka

• Los conceptos de redes relacionados con Managed Service para Apache Kafka.

Roles y permisos obligatorios para crear un clúster

Para obtener los permisos que necesitas para crear un clúster, pídele a tu administrador que te otorgue el rol de IAM Editor de clústeres de Kafka administrados (roles/managedkafka.clusterEditor) en tu proyecto. Para obtener más información sobre cómo otorgar roles, consulta Administra el acceso a proyectos, carpetas y organizaciones.

Este rol predefinido contiene los permisos necesarios para crear un clúster. Para ver los permisos exactos que son necesarios, expande la sección Permisos requeridos:

También puedes obtener estos permisos con roles personalizados o con otros roles predefinidos.

Propiedades de un clúster de Managed Service para Apache Kafka

Cuando creas o actualizas un clúster de Managed Service para Apache Kafka, debes especificar las siguientes propiedades.

Nombre del clúster

Es el nombre o ID del clúster de Servicio administrado para Apache Kafka que estás creando. Si necesitas ayuda para asignarle un nombre a un clúster, consulta los Lineamientos para asignarles nombres a los recursos de Managed Service para Apache Kafka. Los nombres de clústeres son inmutables.

Ubicación

Es la ubicación donde se creará el clúster. La ubicación debe ser una de las regiones admitidas Google Cloud . No podrás cambiarla más adelante. Para ver la lista de ubicaciones disponibles, consulta las ubicaciones de Managed Service para Apache Kafka.

Configuración de capacidad

Capacidad de procesamiento, memoria y almacenamiento del clúster de Kafka. Cuando creas un clúster, debes especificar los siguientes valores:

• vCPUs: Es la cantidad de CPU virtuales en el clúster. Se requieren al menos 3 CPU virtuales por clúster.

• Memoria: Es la cantidad de memoria que se asigna al clúster. Debes aprovisionar entre 1 GiB y 8 GiB por CPU virtual.

  Por ejemplo, si creas un clúster con 6 CPU virtuales, la memoria mínima que puedes asignar al clúster es de 6 GiB (1 GiB por CPU virtual) y la máxima es de 48 GiB (8 GiB por CPU virtual).

El servicio aprovisiona automáticamente agentes, incluidos los volúmenes de Persistent Disk para los agentes. De manera opcional, puedes habilitar Almacenamiento local personalizado, que te permite especificar la capacidad del disco para cada agente. Para obtener más información, consulta Configura el tamaño del disco del agente.

Para obtener instrucciones sobre cómo estimar la capacidad que necesita tu clúster, consulta Planifica el tamaño de tu clúster de Kafka.

Configuración de red

La configuración de red es una lista de subredes de VPC donde el clúster es accesible. Para producir o consumir mensajes, los clientes deben poder acceder a una de estas subredes.

A continuación, se incluyen algunos lineamientos para la configuración de tu red:

• Se requiere un mínimo de 1 subred para un clúster. La cantidad máxima es 10.

• Se permite exactamente 1 subred por red para cualquier clúster determinado.

• Cada subred debe estar en la misma región que el clúster. El proyecto y la red pueden ser diferentes.

• Las direcciones IP del agente y el servidor de arranque se asignan automáticamente en cada subred. Además, las entradas del DNS para estas direcciones IP se crean en las redes de VPC correspondientes.

• Si agregas una subred de un proyecto diferente, debes otorgar permisos a la cuenta de servicio administrada por Google que está asociada al clúster. Para obtener más información, consulta Conecta un clúster entre proyectos.

Después de crear el clúster, puedes actualizar la lista de subredes. Para obtener más información sobre las redes, consulta Configura las redes para Managed Service para Apache Kafka.

Clústeres públicos

De manera opcional, puedes configurar tu clúster como público para permitir que las aplicaciones cliente se conecten a él y ejecuten operaciones de Kafka (incluido el envío y la recepción de mensajes) a través de Internet público. Para habilitar el acceso público, debes especificar uno o más rangos de IP de origen permitidos.

Cuando configuras rangos de IP de origen permitidos, se aplican las siguientes reglas:

• Todos los rangos de IP de origen permitidos deben especificarse en la notación CIDR con IPv4 (por ejemplo, 203.0.113.0/24 o 198.51.100.5/32).

• El tamaño de cada subred de CIDR debe estar entre /16 y /32.

• Los rangos de CIDR no deben superponerse.

• Todos los rangos de IP de origen permitidos deben ser enrutables de forma pública. Se rechazan los rangos de IP privadas, como 10.0.0.0/8 (RFC 1918).

• Puedes especificar un máximo de 500 rangos de IP de origen permitidos.

• No se admiten las direcciones IPv6.

Antes de habilitar el acceso público para un clúster, te recomendamos que leas Cómo conectar clientes a un clúster público para obtener información sobre cómo se implementa la función de acceso público.

Etiquetas

Las etiquetas son pares clave-valor que te ayudan con la organización y la identificación. Las etiquetas te permiten categorizar los recursos según el entorno. Algunos ejemplos son "env:production" y "owner:data-engineering".

Puedes filtrar y buscar recursos según sus etiquetas. Por ejemplo, supongamos que tienes varios clústeres de Managed Service para Apache Kafka para diferentes departamentos. Puedes configurar y buscar clústeres con la etiqueta "department:marketing" para encontrar el que sea pertinente.

Configuración de rebalanceo

Este parámetro de configuración determina si el servicio reequilibra automáticamente las réplicas de particiones entre los intermediarios.

Los modos disponibles son los siguientes:

• Auto rebalance on scale up: Cuando esta opción está habilitada, el servicio activa automáticamente un rebalanceo de las réplicas cuando escalas verticalmente el clúster. Este modo ayuda a mantener una distribución uniforme de la carga, pero es posible que afecte temporalmente el rendimiento durante la operación de rebalanceo.

• No rebalance: Cuando esta opción está habilitada, el servicio no reequilibra automáticamente las réplicas.

Para obtener más información, consulta Cómo reequilibrar las particiones de los agentes.

Encriptación

Managed Service for Apache Kafka puede encriptar mensajes conGoogle-owned and Google-managed encryption keys (predeterminado) o claves de encriptación administradas por el cliente (CMEK). Todos los mensajes se encriptan en reposo y en tránsito. El tipo de encriptación de un clúster es inmutable.

Google-owned and Google-managed encryption keys se usan de forma predeterminada. Google Cloud crea, administra y almacena estas claves por completo dentro de su infraestructura.

Las CMEK son claves de encriptación que administras con Cloud Key Management Service. Esta función te permite tener un mayor control sobre las claves que se usan para encriptar los datos en reposo en los Google Cloud servicios compatibles. El uso de CMEK genera costos adicionales relacionados con Cloud Key Management Service. Para el uso de CMEK, tu llavero de claves debe estar en la misma ubicación que los recursos con los que lo usas. Para obtener más información, consulta Configura la encriptación de mensajes.

Configuración de mTLS

De manera opcional, puedes configurar mTLS como un método de autenticación alternativo que usa certificados de cliente. La configuración incluye lo siguiente:

• Grupos de CA: Es una lista de 1 a 10 grupos de Certificate Authority Service (CAS) en los que confía el clúster para la autenticación del cliente.

• Reglas de asignación de principales de SSL: Es una propiedad del intermediario ssl.principal.mapping.rules opcional, pero recomendada, para simplificar los nombres de principales de certificados largos para su uso en las LCA de Kafka.

Para obtener más información sobre mTLS, consulta Configura la autenticación mTLS.

Crea un clúster

Antes de crear un clúster, lee la documentación de las propiedades del clúster.

Por lo general, la creación de un clúster tarda entre 20 y 30 minutos.

Para crear un clúster, sigue estos pasos:

POST https://managedkafka.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/clusters?clusterId=CLUSTER_ID

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

curl -X POST \
     -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?clusterId=CLUSTER_ID"

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

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

{
  "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": "create",
    "requestedCancellation": false,
    "apiVersion": "v1"
  },
  "done": false
}

resource "google_managed_kafka_cluster" "default" {
  project    = data.google_project.default.project_id # Replace this with your project ID in quotes
  cluster_id = "my-cluster-id"
  location   = "us-central1"
  capacity_config {
    vcpu_count   = 3
    memory_bytes = 3221225472
  }
  gcp_config {
    access_config {
      network_configs {
        subnet = google_compute_subnetwork.default.id
      }
    }
  }
}

import (
	"context"
	"fmt"
	"io"

	"cloud.google.com/go/managedkafka/apiv1/managedkafkapb"
	"google.golang.org/api/option"

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

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

	locationPath := fmt.Sprintf("projects/%s/locations/%s", projectID, region)
	clusterPath := fmt.Sprintf("%s/clusters/%s", locationPath, clusterID)

	// Memory must be between 1 GiB and 8 GiB per CPU.
	capacityConfig := &managedkafkapb.CapacityConfig{
		VcpuCount:   cpu,
		MemoryBytes: memoryBytes,
	}
	var networkConfig []*managedkafkapb.NetworkConfig
	networkConfig = append(networkConfig, &managedkafkapb.NetworkConfig{
		Subnet: subnet,
	})
	platformConfig := &managedkafkapb.Cluster_GcpConfig{
		GcpConfig: &managedkafkapb.GcpConfig{
			AccessConfig: &managedkafkapb.AccessConfig{
				NetworkConfigs: networkConfig,
			},
		},
	}
	rebalanceConfig := &managedkafkapb.RebalanceConfig{
		Mode: managedkafkapb.RebalanceConfig_AUTO_REBALANCE_ON_SCALE_UP,
	}
	cluster := &managedkafkapb.Cluster{
		Name:            clusterPath,
		CapacityConfig:  capacityConfig,
		PlatformConfig:  platformConfig,
		RebalanceConfig: rebalanceConfig,
	}

	req := &managedkafkapb.CreateClusterRequest{
		Parent:    locationPath,
		ClusterId: clusterID,
		Cluster:   cluster,
	}
	op, err := client.CreateCluster(ctx, req)
	if err != nil {
		return fmt.Errorf("client.CreateCluster got err: %w", err)
	}
	// The duration of this operation can vary considerably, typically taking 10-40 minutes.
	resp, err := op.Wait(ctx)
	if err != nil {
		return fmt.Errorf("op.Wait got err: %w", err)
	}
	fmt.Fprintf(w, "Created cluster: %s\n", resp.Name)
	return nil
}

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.RetryingFuture;
import com.google.api.gax.retrying.TimedRetryAlgorithm;
import com.google.cloud.managedkafka.v1.AccessConfig;
import com.google.cloud.managedkafka.v1.CapacityConfig;
import com.google.cloud.managedkafka.v1.Cluster;
import com.google.cloud.managedkafka.v1.CreateClusterRequest;
import com.google.cloud.managedkafka.v1.GcpConfig;
import com.google.cloud.managedkafka.v1.LocationName;
import com.google.cloud.managedkafka.v1.ManagedKafkaClient;
import com.google.cloud.managedkafka.v1.ManagedKafkaSettings;
import com.google.cloud.managedkafka.v1.NetworkConfig;
import com.google.cloud.managedkafka.v1.OperationMetadata;
import com.google.cloud.managedkafka.v1.RebalanceConfig;
import java.time.Duration;
import java.util.concurrent.ExecutionException;

public class CreateCluster {

  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";
    String subnet = "my-subnet"; // e.g. projects/my-project/regions/my-region/subnetworks/my-subnet
    int cpu = 3;
    long memoryBytes = 3221225472L; // 3 GiB
    createCluster(projectId, region, clusterId, subnet, cpu, memoryBytes);
  }

  public static void createCluster(
      String projectId, String region, String clusterId, String subnet, int cpu, long memoryBytes)
      throws Exception {
    CapacityConfig capacityConfig =
        CapacityConfig.newBuilder().setVcpuCount(cpu).setMemoryBytes(memoryBytes).build();
    NetworkConfig networkConfig = NetworkConfig.newBuilder().setSubnet(subnet).build();
    GcpConfig gcpConfig =
        GcpConfig.newBuilder()
            .setAccessConfig(AccessConfig.newBuilder().addNetworkConfigs(networkConfig).build())
            .build();
    RebalanceConfig rebalanceConfig =
        RebalanceConfig.newBuilder()
            .setMode(RebalanceConfig.Mode.AUTO_REBALANCE_ON_SCALE_UP)
            .build();
    Cluster cluster =
        Cluster.newBuilder()
            .setCapacityConfig(capacityConfig)
            .setGcpConfig(gcpConfig)
            .setRebalanceConfig(rebalanceConfig)
            .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.createClusterOperationSettings()
        .setPollingAlgorithm(timedRetryAlgorithm);

    try (ManagedKafkaClient managedKafkaClient = ManagedKafkaClient.create(
        settingsBuilder.build())) {

      CreateClusterRequest request =
          CreateClusterRequest.newBuilder()
              .setParent(LocationName.of(projectId, region).toString())
              .setClusterId(clusterId)
              .setCluster(cluster)
              .build();

      // The duration of this operation can vary considerably, typically taking between 10-40
      // minutes.
      OperationFuture<Cluster, OperationMetadata> future =
          managedKafkaClient.createClusterOperationCallable().futureCall(request);

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

      while (!future.isDone()) {
        // The pollingFuture gives us the most recent status of the operation
        RetryingFuture<OperationSnapshot> pollingFuture = future.getPollingFuture();
        OperationSnapshot currentOp = pollingFuture.getAttemptResult().get();
        System.out.printf("Polling Operation:\nName: %s\n Done: %s\n",
            currentOp.getName(),
            currentOp.isDone());
      }

      // NOTE: future.get() blocks completion until the operation is complete (isDone =  True)
      Cluster response = future.get();
      System.out.printf("Created cluster: %s\n", response.getName());
    } catch (ExecutionException e) {
      System.err.printf("managedKafkaClient.createCluster got err: %s", e.getMessage());
    }
  }
}

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

# TODO(developer)
# project_id = "my-project-id"
# region = "us-central1"
# cluster_id = "my-cluster"
# subnet = "projects/my-project-id/regions/us-central1/subnetworks/default"
# cpu = 3
# memory_bytes = 3221225472

client = managedkafka_v1.ManagedKafkaClient()

cluster = managedkafka_v1.Cluster()
cluster.name = client.cluster_path(project_id, region, cluster_id)
cluster.capacity_config.vcpu_count = cpu
cluster.capacity_config.memory_bytes = memory_bytes
cluster.gcp_config.access_config.network_configs = [
    managedkafka_v1.NetworkConfig(subnet=subnet)
]
cluster.rebalance_config.mode = (
    managedkafka_v1.RebalanceConfig.Mode.AUTO_REBALANCE_ON_SCALE_UP
)

request = managedkafka_v1.CreateClusterRequest(
    parent=client.common_location_path(project_id, region),
    cluster_id=cluster_id,
    cluster=cluster,
)

try:
    operation = client.create_cluster(request=request)
    print(f"Waiting for operation {operation.operation.name} to complete...")
    # The duration of this operation can vary considerably, typically taking 10-40 minutes.
    # We can set a timeout of 3000s (50 minutes).
    response = operation.result(timeout=3000)
    print("Created cluster:", response)
except GoogleAPICallError as e:
    print(f"The operation failed with error: {e.message}")

Supervisa la operación de creación del clúster

Solo puedes ejecutar el siguiente comando si usaste la gcloud CLI para crear el clúster.

• Por lo general, la creación de un clúster tarda entre 20 y 30 minutos. Para hacer un seguimiento del progreso de la creación del clúster, el comando gcloud managed-kafka clusters create usa una operación de larga duración (LRO), que puedes supervisar con el siguiente comando:

  gcloud managed-kafka operations describe OPERATION_ID \
      --location=LOCATION
  

  Reemplaza lo siguiente:

  • OPERATION_ID con el valor del ID de operación de la sección anterior
  • LOCATION con el valor de la ubicación de la sección anterior

Solucionar problemas

Para obtener información sobre cómo solucionar errores relacionados con la creación de un clúster, consulta Errores de creación de clústeres de Kafka.

Próximos pasos

• Supervisa tu clúster.

• Actualiza tu clúster.

• Enumera todos tus clústeres.

• Visualiza tu clúster.

• Borrar un clúster.

• Crea un tema en tu clúster.

• Configura las redes para Managed Service para Apache Kafka.