הצגת רשימה של כל המחברים

רשימה של כל המחברים שפועלים באשכול Connect מספקת סקירה כללית של שילובי הנתונים שהוגדרו. אתם יכולים לעקוב אחרי התקינות והסטטוס של המחברים, לזהות בעיות פוטנציאליות ולנהל את זרימות הנתונים בצורה יעילה.

כדי להציג את כל המחברים באשכול Connect, אפשר להשתמש במסוף Google Cloud , ב-CLI של gcloud, בספריית הלקוח של השירות המנוהל ל-Apache Kafka או ב-Managed Kafka API. אי אפשר להשתמש ב-API של Apache Kafka בקוד פתוח כדי להציג את המחברים.

תפקידים והרשאות שנדרשים כדי להציג רשימה של כל המחברים

כדי לקבל את ההרשאות שנדרשות להצגת רשימה של כל המחברים, צריך לבקש מהאדמין להקצות לכם את תפקיד ה-IAM‏ Managed Kafka Viewer (roles/managedkafka.viewer) בפרויקט. כדי לקרוא הסבר על מתן תפקידים, ראו איך מנהלים את הגישה ברמת הפרויקט, התיקייה והארגון.

זהו תפקיד שמוגדר מראש וכולל את ההרשאות שנדרשות להצגת רשימה של כל המחברים. כדי לראות בדיוק אילו הרשאות נדרשות, אפשר להרחיב את הקטע ההרשאות הנדרשות:

ההרשאות הנדרשות

כדי להציג רשימה של כל המחברים, נדרשות ההרשאות הבאות:

• נותנים את ההרשאה list connectors באשכול Connect הראשי: managedkafka.connectors.list
• נותנים את ההרשאה get connector details (קבלת פרטי המחבר) באשכול הראשי של Connect: managedkafka.connectors.get

יכול להיות שתקבלו את ההרשאות האלה באמצעות תפקידים בהתאמה אישית או תפקידים מוגדרים מראש אחרים.

לצפייה בכל המחברים

התצוגה הזו מאפשרת לעקוב במהירות אחרי הסטטוס של מחברי הנתונים ולזהות מחברים שדורשים התייחסות. אחר כך תוכלו להתעמק במחברים ספציפיים כדי לראות את הפרטים וההגדרות שלהם, אם צריך.

המסוף

1. נכנסים לדף Connect clusters במסוף Google Cloud .

   מעבר אל Connect Clusters

2. לוחצים על Connect cluster (אשכול Connect) שרוצים להציג את המחברים שלו.

   יוצג הדף Connect cluster details.

3. בכרטיסייה Resources מוצגת רשימה של כל המחברים שפועלים באשכול. הרשימה כוללת את הפרטים הבאים לגבי כל מחבר:

   • שם: השם של המחבר.
   • מצב: מצב הפעולה של המחבר. לדוגמה, פועל, נכשל.
   • סוג המחבר: סוג תוסף המחבר.

   אפשר להשתמש באפשרות סינון כדי לחפש מחברים ספציפיים לפי שם.

gcloud

1. במסוף Google Cloud , מפעילים את Cloud Shell.

   הפעלת Cloud Shell

   בחלק התחתון של Google Cloud המסוף יתחיל סשן של Cloud Shell ותופיע הודעה של שורת הפקודה. Cloud Shell היא סביבת מעטפת שבה ה-CLI של Google Cloud מותקן ומוגדרים ערכים לפרויקט הקיים. הסשן יופעל תוך כמה שניות.

2. משתמשים בפקודה gcloud managed-kafka connectors list כדי להציג את רשימת המחברים:

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

3. כדי לצמצם עוד יותר את רשימת המחברים, אפשר להשתמש בדגלים נוספים:

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

   מחליפים את מה שכתוב בשדות הבאים:

   • ‫CONNECT_CLUSTER_ID: שדה חובה. המזהה של אשכול Connect שמכיל את המחברים שרוצים להציג ברשימה.
   • ‫LOCATION: שדה חובה. המיקום של Connect cluster שמכיל את המחברים שרוצים להציג.

   • ‫EXPRESSION: (אופציונלי) ביטוי של מסנן בוליאני שיוחל על הרשימה. אם הערך של הביטוי הוא True, הפריט ייכלל ברשימה. לפרטים נוספים ולדוגמאות, מריצים את הפקודה gcloud topic filters.

     דוגמאות:

     • כדי להציג רק מחברים במצב 'פועל':

       --filter="state=RUNNING"
       

     • כדי להציג רק מחברים מסוג Pub/Sub Sink:

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

     • כדי להציג רשימה של מחברים שהשם שלהם מכיל את המחרוזת prod:

       --filter="name ~ 'prod'"
       

     • כדי להציג רשימה של מחברים עם הערך FAILED או עם הפלאגין Pub/Sub Source:

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

   • ‫LIMIT: (אופציונלי) המספר המקסימלי של מחברים שיוצגו. אם לא מציינים מחברים, מוצגים כל המחברים.

   • ‫PAGE_SIZE: (אופציונלי) מספר התוצאות שיוצגו בכל דף. אם לא מציינים ערך, השירות קובע גודל דף מתאים.

   • ‫SORT_BY: (אופציונלי) רשימה מופרדת בפסיקים של שדות שלפיהם רשימת המשאבים תמוין. סדר המיון שמוגדר כברירת מחדל הוא סדר עולה. כדי למיין בסדר יורד, מוסיפים את התו ~ בתחילת השדה. השדות הנתמכים הם כנראה name ו-state.

פקודה לדוגמה עם מיון:

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

דוגמה לפקודה עם סינון והגבלה:

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

פלט לדוגמה:

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 במאמר התקנת ספריות הלקוח. מידע נוסף מופיע ב מאמרי העזרה של ה-API של שירות מנוהל ל-Apache Kafka Go.

כדי לבצע אימות לשירות המנוהל ל-Apache Kafka, צריך להגדיר את Application Default Credentials‏(ADC). מידע נוסף זמין במאמר הגדרת ADC לסביבת פיתוח מקומית.

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

לפני שמנסים להריץ את הדוגמה הזו, צריך לפעול לפי הוראות ההגדרה של Java במאמר התקנת ספריות הלקוח. מידע נוסף מופיע ב מאמרי העזרה של Managed Service for Apache Kafka Java API.

כדי לבצע אימות לשירות המנוהל ל-Apache Kafka, מגדירים את ה-Application Default Credentials. מידע נוסף זמין במאמר הגדרת ADC לסביבת פיתוח מקומית.

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

לפני שמנסים להריץ את הדוגמה הזו, צריך לפעול לפי הוראות ההגדרה של Python במאמר התקנת ספריות הלקוח. מידע נוסף מופיע ב מאמרי העזרה של Managed Service for Apache Kafka Python API.

כדי לבצע אימות לשירות המנוהל ל-Apache Kafka, מגדירים את ה-Application Default Credentials. מידע נוסף זמין במאמר הגדרת ADC לסביבת פיתוח מקומית.

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®‎ הוא סימן מסחרי רשום של The Apache Software Foundation או של השותפים העצמאיים שלה בארצות הברית או במדינות אחרות.