Enumera todos los conectores

Enumerar todos los conectores que se ejecutan en un clúster de Connect proporciona una descripción general de las integraciones de datos que están configuradas. Puedes supervisar el estado de tus conectores, identificar posibles problemas y administrar tus flujos de datos de manera eficaz.

Para enumerar todos los conectores en un clúster de Connect, puedes usar la Google Cloud consola, la gcloud CLI, la biblioteca cliente de Managed Service para Apache Kafka o la API de Managed Kafka. No puedes usar la API de Apache Kafka de código abierto para enumerar los conectores.

Roles y permisos obligatorios para enumerar todos los conectores

Para obtener los permisos que necesitas para enumerar todos los conectores, pídele a tu administrador que te otorgue el rol de IAM de Visualizador de Kafka administrado (roles/managedkafka.viewer) 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 enumerar todos los conectores. Para ver los permisos exactos que son necesarios, expande la sección Permisos requeridos:

Permisos necesarios

Se requieren los siguientes permisos para enumerar todos los conectores:

  • Otorga el permiso de conectores de lista en el clúster de conexión principal: managedkafka.connectors.list
  • Otorga el permiso para obtener detalles del conector en el clúster principal de Connect: managedkafka.connectors.get

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

Ver todos los conectores

Esta vista proporciona una forma rápida de supervisar el estado de tus conectores e identificar los que requieren atención. Luego, puedes desglosar los conectores individuales para ver sus detalles y configuraciones si es necesario.

Console

  1. En la consola de Google Cloud , ve a la página Conectar clústeres.

    Ir a Connect Clusters

  2. Haz clic en el clúster de Connect para el que deseas enumerar los conectores.

    Se muestra la página Detalles de conexión del clúster.

  3. En la pestaña Resources, se muestra una lista de todos los conectores que se ejecutan en el clúster. La lista incluye la siguiente información para cada conector:

    • Nombre: Es el nombre del conector.
    • Estado: Es el estado operativo del conector. Por ejemplo, en ejecución, falló.
    • Tipo de conector: Es el tipo de complemento del conector.

    Puedes usar la opción Filtrar para buscar conectores específicos por nombre.

gcloud

  1. En la consola de Google Cloud , activa Cloud Shell.

    Activa Cloud Shell

    En la parte inferior de la consola de Google Cloud , se inicia una sesión de Cloud Shell en la que se muestra una ventana de línea de comandos. Cloud Shell es un entorno de shell con Google Cloud CLI ya instalada y con valores ya establecidos para el proyecto actual. La sesión puede tardar unos segundos en inicializarse.

  2. Usa el comando gcloud managed-kafka connectors list para enumerar los conectores:

    gcloud managed-kafka connectors list CONNECT_CLUSTER_ID \
    --location=LOCATION

  3. Para definir mejor la lista de conectores, puedes usar marcas adicionales:

    gcloud managed-kafka connectors list CONNECT_CLUSTER_ID \
    --location=LOCATION \
    [--filter=EXPRESSION] \
    [--limit=LIMIT] \
    [--page-size=PAGE_SIZE] \
    [--sort-by=SORT_BY]

    Reemplaza lo siguiente:

    • CONNECT_CLUSTER_ID: Obligatorio. Es el ID del clúster de Connect que contiene los conectores que deseas enumerar.
    • LOCATION: Obligatorio. Ubicación del clúster de Connect que contiene los conectores que deseas enumerar.

    • EXPRESSION: (Opcional) Es una expresión de filtro booleano que se aplicará a la lista. Si la expresión se evalúa como True, el elemento se incluye en la lista. Para obtener más detalles y ejemplos, ejecuta gcloud topic filters.

      Ejemplos:

      • Para enumerar solo los conectores que están en el estado "RUNNING", haz lo siguiente:

        --filter="state=RUNNING"

      • Para enumerar solo los conectores de "receptor de Pub/Sub", haz lo siguiente:

        --filter="connector_plugin='Pub/Sub Sink'"

      • Para enumerar los conectores con un nombre que contenga "prod", haz lo siguiente:

        --filter="name ~ 'prod'"

      • Para enumerar los conectores que tienen el estado "FAILED" o son complementos de "Pub/Sub Source", haz lo siguiente:

        --filter="state=FAILED OR connector_plugin='Pub/Sub Source'"

    • LIMIT: (Opcional) Es la cantidad máxima de conectores que se mostrarán. Si no se especifica, se muestran todos los conectores.

    • PAGE_SIZE: (Opcional) Es la cantidad de resultados que se mostrarán por página. Si no se especifica, el servicio determina un tamaño de página adecuado.

    • SORT_BY: (Opcional) Lista separada por comas de campos según los cuales se ordenará la respuesta. El orden de clasificación predeterminado es ascendente. Para ordenar de forma descendente, agrega el prefijo ~ al campo. Los campos admitidos suelen ser name y state.

Comando de ejemplo con ordenamiento:

gcloud managed-kafka connectors list test-connect-cluster \
    --location=us-central1 \
    --sort-by=~state,name

Comando de ejemplo con filtrado y limitación:

gcloud managed-kafka connectors list test-connect-cluster \
    --location=us-central1 \
    --filter="state=RUNNING AND connector_plugin='Pub/Sub Sink'" \
    --limit=5

Resultado de ejemplo:

NAME                                    STATE     CONNECTOR_PLUGIN
pubsub-sink-connector                   RUNNING   Pub/Sub Sink
another-pubsub-sink                     RUNNING   Pub/Sub Sink
prod-pubsub-sink                        RUNNING   Pub/Sub Sink

Go

Antes de probar este ejemplo, sigue las instrucciones de configuración para Go en Instala las bibliotecas cliente. Si deseas obtener más información, consulta la documentación de referencia de la API de Go de Managed Service para Apache Kafka.

Para autenticarte en Managed Service para Apache Kafka, configura las credenciales predeterminadas de la aplicación(ADC). Para obtener más información, consulta Configura ADC para un entorno de desarrollo local.

import (
	"context"
	"fmt"
	"io"

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

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

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

	parent := fmt.Sprintf("projects/%s/locations/%s/connectClusters/%s", projectID, region, connectClusterID)
	req := &managedkafkapb.ListConnectorsRequest{
		Parent: parent,
	}
	connectorIter := client.ListConnectors(ctx, req)
	for {
		connector, err := connectorIter.Next()
		if err == iterator.Done {
			break
		}
		if err != nil {
			return fmt.Errorf("connectorIter.Next() got err: %w", err)
		}
		fmt.Fprintf(w, "Got connector: %v", connector)
	}
	return nil
}

Java

Antes de probar este ejemplo, sigue las instrucciones de configuración para Java que se encuentran en Cómo instalar las bibliotecas cliente. Para obtener más información, consulta la documentación de referencia de la API de Managed Service para Apache Kafka en Java.

Para autenticarte en Managed Service para Apache Kafka, configura las credenciales predeterminadas de la aplicación. Para obtener más información, consulta Configura ADC para un entorno de desarrollo local.

import com.google.api.gax.rpc.ApiException;
import com.google.cloud.managedkafka.v1.ConnectClusterName;
import com.google.cloud.managedkafka.v1.Connector;
import com.google.cloud.managedkafka.v1.ManagedKafkaConnectClient;
import java.io.IOException;

public class ListConnectors {

  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-connect-cluster";
    listConnectors(projectId, region, clusterId);
  }

  public static void listConnectors(String projectId, String region, String clusterId)
      throws IOException {
    try (ManagedKafkaConnectClient managedKafkaConnectClient = ManagedKafkaConnectClient.create()) {
      ConnectClusterName parent = ConnectClusterName.of(projectId, region, clusterId);
      // This operation is handled synchronously.
      for (Connector connector : managedKafkaConnectClient.listConnectors(parent).iterateAll()) {
        System.out.println(connector.getAllFields());
      }
    } catch (IOException | ApiException e) {
      System.err.printf("managedKafkaConnectClient.listConnectors got err: %s\n", e.getMessage());
    }
  }
}

Python

Antes de probar este ejemplo, sigue las instrucciones de configuración para Python que se encuentran en Instala las bibliotecas cliente. Si deseas obtener más información, consulta la documentación de referencia de la API de Python de Managed Service para Apache Kafka.

Para autenticarte en Managed Service para Apache Kafka, configura las credenciales predeterminadas de la aplicación. Para obtener más información, consulta Configura ADC para un entorno de desarrollo local.

from google.cloud import managedkafka_v1
from google.cloud.managedkafka_v1.services.managed_kafka_connect import (
    ManagedKafkaConnectClient,
)
from google.api_core.exceptions import GoogleAPICallError

# TODO(developer)
# project_id = "my-project-id"
# region = "us-central1"
# connect_cluster_id = "my-connect-cluster"

connect_client = ManagedKafkaConnectClient()

request = managedkafka_v1.ListConnectorsRequest(
    parent=connect_client.connect_cluster_path(project_id, region, connect_cluster_id),
)

try:
    response = connect_client.list_connectors(request=request)
    for connector in response:
        print("Got connector:", connector)
except GoogleAPICallError as e:
    print(f"Failed to list connectors with error: {e}")

Apache Kafka® es una marca registrada de The Apache Software Foundation o sus afiliados en Estados Unidos y otros países.