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

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

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

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

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

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

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

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

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

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

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

המסוף

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

   מעבר אל Clusters

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

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

gcloud

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

   הפעלת Cloud Shell

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

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

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

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

   • ‫LOCATION: המיקום של האשכול.
   • ‫LIMIT: המספר המקסימלי של האשכולות שיופיעו ברשימה.

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 של שירות מנוהל ל-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 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 או של השותפים העצמאיים שלה בארצות הברית או במדינות אחרות.