הצגת רשימה של אשכולות בשירות המנוהל של Google Cloud ל-Apache Kafka

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

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

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

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

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

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

• רשימת האשכולות: managedkafka.clusters.list
• מציגים את פרטי האשכול: managedkafka.clusters.get

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

הצגת רשימת האשכולות

כדי להציג את כל האשכולות בפרויקט:

המסוף

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

   כניסה לדף Clusters

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

   • ‫Cluster name: המזהה הייחודי של האשכול בשירות המנוהל ל-Apache Kafka. אפשר להשתמש בו כדי להפנות לאשכול בפעולות שונות.
   • ‫Cluster state: סטטוס הפעילות הנוכחי של האשכול, למשל Active.
   • ‫Region: המיקום הגיאוגרפי שבו מארחים את האשכול.
   • כתובת URL התחלתית: הכתובת ההתחלתית שבה משתמשים לקוחות Kafka כדי להתחבר לאשכול.
   • זיכרון: נפח הזיכרון הכולל שהוקצה לאשכול. הוא קובע את הקיבולת לטיפול בתנועת ההודעות ובאחסון.
   • ‫vCPU: מספר המעבדים הווירטואליים שהוקצו לאשכול. זה משפיע על כוח העיבוד של האשכול ועל היכולת שלו לטפל בפעולות בו-זמניות.
   • ‫Labels: צמדי מפתח/ערך שאפשר לצרף לאשכול למטרות ארגון, סינון ואוטומציה.

3. אופציונלי: כדי למצוא את האשכול שמשויך לכתובת התחלתית ספציפית, מזינים את כתובת ה-URL ההתחלתית בשדה Filter.

gcloud

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

   הפעלת Cloud Shell

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

2. מריצים את הפקודה gcloud managed-kafka clusters list:

   gcloud managed-kafka clusters list --location=LOCATION \
       --limit=LIMIT
   

   כדי למצוא את האשכול שמשויך לכתובת אתחול ספציפית, מסננים לפי השדה bootstrapAddress:

   gcloud managed-kafka clusters list --location=LOCATION \
       --filter="bootstrapAddress:BOOTSTRAP_ADDRESS"
   

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

   • ‫LOCATION: המיקום (האזור) של האשכול, למשל us-central1. האזור נכלל גם בתור הקטע השלישי שמופרד בנקודה בכתובת ה-URL ההתחלתית.
   • ‫LIMIT: המספר המקסימלי של האשכולות שיופיעו ברשימה.
   • ‫BOOTSTRAP_ADDRESS: כתובת ה-URL ההתחלתית או שם המארח של האשכול.

REST

לפני שמשתמשים בנתוני הבקשה, צריך להחליף את הנתונים הבאים:

• ‫PROJECT_ID: מזהה הפרויקט ב- Google Cloud
• ‫LOCATION: המיקום של האשכול

ה-method של ה-HTTP וכתובת ה-URL:

GET https://managedkafka.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/clusters

כדי לשלוח את הבקשה צריך להרחיב אחת מהאפשרויות הבאות:

‫Curl (Linux,‏ macOS או Cloud Shell)

הערה: הפקודה הבאה מבוססת על ההנחה שנכנסתם ל-CLI של gcloud באמצעות חשבון המשתמש שלכם, על ידי הרצת gcloud init או gcloud auth login, או באמצעות Cloud Shell שמחבר אתכם אוטומטית ל-CLI של gcloud. כדי לבדוק איזה חשבון פעיל, אפשר להריץ את הפקודה gcloud auth list.

מריצים את הפקודה הבאה:

curl -X GET \
     -H "Authorization: Bearer $(gcloud auth print-access-token)" \
     "https://managedkafka.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/clusters"

‎PowerShell (Windows)

הערה: הפקודה הבאה מבוססת על ההנחה שנכנסתם ל-CLI של gcloud באמצעות חשבון המשתמש שלכם, על ידי הרצת gcloud init או gcloud auth login. כדי לבדוק איזה חשבון פעיל, אפשר להריץ את הפקודה gcloud auth list.

מריצים את הפקודה הבאה:

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

Invoke-WebRequest `
    -Method GET `
    -Headers $headers `
    -Uri "https://managedkafka.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/clusters" | Select-Object -Expand Content

אתם אמורים לקבל תגובת JSON שדומה לזו:

{
  "clusters": [
    {
      "name": "projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID",
      "createTime": "CREATE_TIME",
      "updateTime": "UPDATE_TIME",
      "capacityConfig": {
        "vcpuCount": "CPU_COUNT",
        "memoryBytes": "MEMORY"
      },
      "rebalanceConfig": {},
      "gcpConfig": {
        "accessConfig": {
          "networkConfigs": [
            {
              "subnet": "projects/PROJECT_ID/locations/LOCATION/subnetworks/SUBNET_ID"
            }
          ]
        }
      },
      "state": "ACTIVE",
      "satisfiesPzi": false,
      "satisfiesPzs": false,
      "tlsConfig": {
        "trustConfig": {}
      },
      "updateOptions": {}
    }
  ]
}

המשך

לפני שמנסים את הדוגמה הזו, צריך לפעול לפי הוראות ההגדרה של Go בקטע התקנת ספריות הלקוח. מידע נוסף מופיע ב מאמרי העזרה של ה-API של Go עבור שירות מנוהל ל-Apache Kafka.

כדי לבצע אימות בשירות המנוהל ל-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 listClusters(w io.Writer, projectID, region string, opts ...option.ClientOption) error {
	// projectID := "my-project-id"
	// region := "us-central1"
	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)
	req := &managedkafkapb.ListClustersRequest{
		Parent: locationPath,
	}
	clusterIter := client.ListClusters(ctx, req)
	for {
		res, err := clusterIter.Next()
		if err == iterator.Done {
			break
		}
		if err != nil {
			return fmt.Errorf("clusterIter.Next() got err: %w", err)
		}
		fmt.Fprintf(w, "Got cluster: %v", res)
	}
	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.Cluster;
import com.google.cloud.managedkafka.v1.LocationName;
import com.google.cloud.managedkafka.v1.ManagedKafkaClient;
import java.io.IOException;

public class ListClusters {

  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
    listClusters(projectId, region);
  }

  public static void listClusters(String projectId, String region) throws Exception {
    try (ManagedKafkaClient managedKafkaClient = ManagedKafkaClient.create()) {
      LocationName locationName = LocationName.of(projectId, region);
      // This operation is being handled synchronously.
      for (Cluster cluster : managedKafkaClient.listClusters(locationName).iterateAll()) {
        System.out.println(cluster.getAllFields());
      }
    } catch (IOException | ApiException e) {
      System.err.printf("managedKafkaClient.listClusters got err: %s", e.getMessage());
    }
  }
}

Python

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

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

from google.cloud import managedkafka_v1

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

client = managedkafka_v1.ManagedKafkaClient()

request = managedkafka_v1.ListClustersRequest(
    parent=client.common_location_path(project_id, region),
)

response = client.list_clusters(request=request)
for cluster in response:
    print("Got cluster:", cluster)

מה השלב הבא?

• מעקב אחרי האשכול.

• הצגת האשכול.

• יוצרים נושא באשכול.

• עריכת אשכול.

• מחיקת האשכול.

‫Apache Kafka®‎ הוא סימן מסחרי רשום של The Apache Software Foundation או של השותפים העצמאיים שלה בארצות הברית או במדינות אחרות.