עדכון של אשכול בשירות מנוהל של Google Cloud ל-Apache Kafka

אפשר לערוך אשכול של שירות מנוהל של Google Cloud ל-Apache Kafka כדי לעדכן מאפיינים כמו גודל האשכול, כולל מספר ליבות ה-vCPU והזיכרון, רשימת רשתות המשנה המחוברות, טווחי כתובות ה-IP המותרים של המקור לאשכולות ציבוריים, הגדרת האיזון מחדש האוטומטי והגדרת ה-mTLS.

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

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

תפקידים והרשאות נדרשים

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

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

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

כדי לעדכן אשכול, צריך את ההרשאות הבאות:

  • עריכת אשכול: managedkafka.clusters.update

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

שינוי הגודל של אוסף

אם מעדכנים את מספר ליבות ה-vCPU או את הזיכרון של אשכול, הכללים הבאים חלים:

  • יחס ה-vCPU לזיכרון הכולל של האשכול חייב תמיד להיות בין 1:1 ל-1:8.

  • צריך להקצות לפחות 1 vCPU ו-1 GiB של זיכרון לכל ברוקר קיים. מספר הברוקרים אף פעם לא יורד.

  • אם באשכול יש הגדרת דיסק בהתאמה אישית, העדכון צריך לעמוד בדרישות להגדרת הדיסק לאחסון מקומי.

  • אם משדרגים את הקיבולת, הממוצע של vCPU וזיכרון לכל ברוקר לא יכול לרדת ביותר מ-10% בהשוואה לממוצעים לפני העדכון. לדוגמה, אם מנסים להגדיל את הקיבולת של אשכול מ-45 ליבות וירטואליות (3 ברוקרים) ל-48 ליבות וירטואליות (4 ברוקרים), ממוצע הליבות הווירטואליות לכל ברוקר יורד מ-15 ל-12, כלומר ירידה של 20%, שחורגת מהמגבלה של 10%.

    אם אתם צריכים להקטין את מספר ליבות ה-vCPU ביותר מ-10%, מומלץ להקטין אותו בכמה שלבים. אחרי כל עדכון, עוקבים אחרי ניצול המשאבים ומאזנים מחדש את המחיצות אם צריך.

    עם זאת, אם אתם בטוחים שלברוקרים שלכם יהיה מספיק קיבולת אחרי העדכון, אתם יכולים להשבית את הבדיקה הזו על ידי הפעלת הפקודה gcloud managed-kafka clusters update עם הדגל allow_broker_downscale_on_cluster_upscale=true. הסימן הזה מציין שאתם מקבלים את הסיכון הפוטנציאלי לביצועים.

מידע נוסף זמין במאמר בנושא עדכון גודל האשכול.

הגדרת אשכול ציבורי

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

בשירות המנוהל ל-Apache Kafka נעשה שימוש ב-Cloud Next Generation Firewall כדי להגביל את הגישה לאשכולות ציבוריים. הסרה של טווחי כתובות IP מותרים או השבתה של גישה ציבורית חלות רק על חיבורים חדשים. מידע נוסף זמין במאמר בנושא השפעות על תנועה קיימת.

עריכת אשכול

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

המסוף

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

מעבר אל Clusters

  1. ברשימת האשכולות, לוחצים על האשכול שרוצים לערוך את המאפיינים שלו.

במסוף מוצג דף הפרטים של האשכול.

  1. בדף הפרטים של האשכול, לוחצים על עריכה.

  2. עורכים את המאפיינים לפי הצורך. אפשר לערוך את המאפיינים הבאים של אשכול במסוף:

    • זיכרון
    • מעבדים וירטואלים
    • תת-רשת
    • הגדרות האיזון מחדש
    • הגדרת mTLS
    • תוויות

  3. לוחצים על Save.

gcloud

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

    הפעלת Cloud Shell

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

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

    • ‫PROJECT_ID: מזהה הפרויקט.
    • ‫LOCATION: המיקום של האשכול.
    • ‫CLUSTER_ID: מזהה האשכול.
    • ‫CPU_COUNT: מספר ליבות ה-vCPU באשכול.
    • ‫MEMORY: כמות הזיכרון של האשכול. לדוגמה: 10GiB.
    • ‫SUBNET_ID: מזהה תת-הרשת שאליה רוצים להתחבר. לדוגמה: default.
    • ‫LABELS: התוויות לשיוך לאשכול.
    • ‫ALLOWED_SOURCE_IP_RANGES: טווחי ה-CIDR המורשים של IPv4 לגישה לאינטרנט באשכול ציבורי.

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

    ‫Linux,‏ macOS או Cloud Shell

    gcloud managed-kafka clusters update CLUSTER_ID \
    --location=LOCATION \
    --cpu=CPU_COUNT \
    --memory=MEMORY \
    --subnets=projects/PROJECT_ID/regions/LOCATION/subnetworks/SUBNET_ID \
    --auto-rebalance \
    --labels=LABELS \
    --public-cluster \
    --allowed-source-ip-ranges=ALLOWED_SOURCE_IP_RANGES

    ‏Windows (PowerShell)

    gcloud managed-kafka clusters update CLUSTER_ID `
    --location=LOCATION `
    --cpu=CPU_COUNT `
    --memory=MEMORY `
    --subnets=projects/PROJECT_ID/regions/LOCATION/subnetworks/SUBNET_ID `
    --auto-rebalance `
    --labels=LABELS `
    --public-cluster `
    --allowed-source-ip-ranges=ALLOWED_SOURCE_IP_RANGES

    Windows‏ (cmd.exe)

    gcloud managed-kafka clusters update CLUSTER_ID ^
    --location=LOCATION ^
    --cpu=CPU_COUNT ^
    --memory=MEMORY ^
    --subnets=projects/PROJECT_ID/regions/LOCATION/subnetworks/SUBNET_ID ^
    --auto-rebalance ^
    --labels=LABELS ^
    --public-cluster ^
    --allowed-source-ip-ranges=ALLOWED_SOURCE_IP_RANGES

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

    done: false
metadata:
  '@type': type.googleapis.com/google.cloud.managedkafka.v1.OperationMetadata
  apiVersion: v1
  createTime: 'CREATE_TIME'
  requestedCancellation: false
  target: projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID
  verb: update
name: projects/PROJECT_ID/locations/LOCATION/operations/OPERATION_ID
    • כדי להשבית את הגישה הציבורית, משתמשים בדגל --no-public-cluster.
    • אם משתמשים בדגל --async עם הפקודה, המערכת שולחת את בקשת העדכון ומחזירה תגובה באופן מיידי, בלי לחכות לסיום הפעולה. בעזרת הדגל --async, אפשר להמשיך במשימות אחרות בזמן שעדכון האשכול מתבצע ברקע. אם לא משתמשים בדגל --async, המערכת מחכה שהפעולה תושלם לפני שהיא מחזירה תשובה. צריך לחכות עד שהעדכון של האשכול יסתיים לפני שממשיכים במשימות אחרות.

REST

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

  • ‫PROJECT_ID: מזהה הפרויקט ב- Google Cloud
  • ‫LOCATION: המיקום של האשכול.
  • ‫CLUSTER_ID: מזהה האשכול.
  • ‫UPDATE_MASK: השדות לעדכון, כרשימה מופרדת בפסיקים של שמות מלאים. דוגמה: capacityConfig.vcpuCount,capacityConfig.memoryBytes
  • ‫CPU_COUNT: מספר ליבות ה-vCPU באשכול.
  • ‫MEMORY: כמות הזיכרון באשכול, בבייטים. לדוגמה: 3221225472.
  • ‫SUBNET_ID: מזהה תת-הרשת שאליה רוצים להתחבר. לדוגמה: default.

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

PATCH https://managedkafka.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID?updateMask=UPDATE_MASK

תוכן בקשת JSON: 

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

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

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

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

בגוף הבקשה, כוללים רק את השדות שרוצים לעדכן, כפי שמצוין בפרמטר השאילתה UPDATE_MASK.

  • כדי להוסיף תת-רשת, מוסיפים רשומה חדשה ל-networkConfigs בפורמט הבא: projects/PROJECT_ID/regions/LOCATION/subnetworks/SUBNET_ID. לדוגמה: projects/sample-project/regions/us-central1/subnetworks/default.

  • כדי להפעיל גישה ציבורית או לעדכן את טווחי כתובות ה-IP המותרים של המקור, צריך לכלול את gcpConfig.accessConfig.publicClusterConfig בפרמטר השאילתה UPDATE_MASK ולציין את מערך allowedSourceIpRanges בגוף הבקשה. גוף בקשה לדוגמה:

    {
  "gcpConfig": {
    "accessConfig": {
      "publicClusterConfig": {
        "allowedSourceIpRanges": [
          "203.0.113.0/24"
        ]
      }
    }
  }
}

  • כדי להשבית את הגישה הציבורית, כוללים את gcpConfig.accessConfig.publicClusterConfig בפרמטר השאילתה UPDATE_MASK ומעבירים אובייקט JSON ריק {} בגוף הבקשה (או משמיטים את publicClusterConfig). גוף הבקשה לדוגמה:

    {}

המשך

לפני שמנסים להריץ את הדוגמה הזו, צריך לפעול לפי הוראות ההגדרה של 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/option"
	"google.golang.org/protobuf/types/known/fieldmaskpb"

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

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

	clusterPath := fmt.Sprintf("projects/%s/locations/%s/clusters/%s", projectID, region, clusterID)
	capacityConfig := &managedkafkapb.CapacityConfig{
		MemoryBytes: memory,
	}
	cluster := &managedkafkapb.Cluster{
		Name:           clusterPath,
		CapacityConfig: capacityConfig,
	}
	paths := []string{"capacity_config.memory_bytes"}
	updateMask := &fieldmaskpb.FieldMask{
		Paths: paths,
	}

	req := &managedkafkapb.UpdateClusterRequest{
		UpdateMask: updateMask,
		Cluster:    cluster,
	}
	op, err := client.UpdateCluster(ctx, req)
	if err != nil {
		return fmt.Errorf("client.UpdateCluster got err: %w", err)
	}
	resp, err := op.Wait(ctx)
	if err != nil {
		return fmt.Errorf("op.Wait got err: %w", err)
	}
	fmt.Fprintf(w, "Updated cluster: %#v\n", resp)
	return nil
}

Java

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

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


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.TimedRetryAlgorithm;
import com.google.cloud.managedkafka.v1.CapacityConfig;
import com.google.cloud.managedkafka.v1.Cluster;
import com.google.cloud.managedkafka.v1.ClusterName;
import com.google.cloud.managedkafka.v1.ManagedKafkaClient;
import com.google.cloud.managedkafka.v1.ManagedKafkaSettings;
import com.google.cloud.managedkafka.v1.OperationMetadata;
import com.google.cloud.managedkafka.v1.UpdateClusterRequest;
import com.google.protobuf.FieldMask;
import java.time.Duration;
import java.util.concurrent.ExecutionException;

public class UpdateCluster {

  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";
    long memoryBytes = 25769803776L; // 24 GiB
    updateCluster(projectId, region, clusterId, memoryBytes);
  }

  public static void updateCluster(
      String projectId, String region, String clusterId, long memoryBytes) throws Exception {
    CapacityConfig capacityConfig = CapacityConfig.newBuilder().setMemoryBytes(memoryBytes).build();
    Cluster cluster =
        Cluster.newBuilder()
            .setName(ClusterName.of(projectId, region, clusterId).toString())
            .setCapacityConfig(capacityConfig)
            .build();
    FieldMask updateMask = FieldMask.newBuilder().addPaths("capacity_config.memory_bytes").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.updateClusterOperationSettings()
        .setPollingAlgorithm(timedRetryAlgorithm);

    try (ManagedKafkaClient managedKafkaClient = ManagedKafkaClient.create(
        settingsBuilder.build())) {
      UpdateClusterRequest request =
          UpdateClusterRequest.newBuilder().setUpdateMask(updateMask).setCluster(cluster).build();
      OperationFuture<Cluster, OperationMetadata> future =
          managedKafkaClient.updateClusterOperationCallable().futureCall(request);

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

      Cluster response = future.get();
      System.out.printf("Updated cluster: %s\n", response.getName());
    } catch (ExecutionException e) {
      System.err.printf("managedKafkaClient.updateCluster got err: %s", e.getMessage());
    }
  }
}

Python

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

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

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

# TODO(developer)
# project_id = "my-project-id"
# region = "us-central1"
# cluster_id = "my-cluster"
# memory_bytes = 4295000000

client = managedkafka_v1.ManagedKafkaClient()

cluster = managedkafka_v1.Cluster()
cluster.name = client.cluster_path(project_id, region, cluster_id)
cluster.capacity_config.memory_bytes = memory_bytes
update_mask = field_mask_pb2.FieldMask()
update_mask.paths.append("capacity_config.memory_bytes")

# For a list of editable fields, one can check https://cloud.google.com/managed-kafka/docs/create-cluster#properties.
request = managedkafka_v1.UpdateClusterRequest(
    update_mask=update_mask,
    cluster=cluster,
)

try:
    operation = client.update_cluster(request=request)
    print(f"Waiting for operation {operation.operation.name} to complete...")
    response = operation.result()
    print("Updated cluster:", response)
except GoogleAPICallError as e:
    print(f"The operation failed with error: {e.message}")

מגבלות

אחרי שיוצרים אשכול של שירות מנוהל ל-Apache Kafka, אי אפשר לעדכן את המאפיינים הבאים:

  • שם האשכול
  • המיקום של האשכול
  • סוג ההצפנה

אי אפשר לשנות את סוג ההצפנה, אבל אפשר לסובב מפתחות הצפנה.

מה השלב הבא?

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