פריסת Apache Kafka ב-GKE באמצעות Confluent

טייס אוטומטי רגילה

במדריך הזה נסביר איך להשתמש באופרטור Confluent for Kubernetes (CFK) כדי לפרוס אשכולות של Apache Kafka ב-Google Kubernetes Engine ‏ (GKE).

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

המדריך הזה מיועד לאדמינים של פלטפורמות, למומחי Cloud Architect ולמומחי תפעול שרוצים לפרוס אשכולות Kafka ב-GKE.

אפשר גם להשתמש באופרטור CFK כדי לפרוס רכיבים אחרים של Confluent Platform, כמו Confluent Control Center מבוסס-האינטרנט, Schema Registry או KsqlDB. עם זאת, המדריך הזה מתמקד רק בפריסות של Kafka.

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

מטרות

• תכנון ופריסה של תשתית GKE ל-Apache Kafka
• פריסה והגדרה של אופרטור CFK
• הגדרת Apache Kafka באמצעות אופרטור CFK כדי להבטיח זמינות, אבטחה, יכולת צפייה וביצועים

יתרונות

היתרונות של CFK:

• עדכונים אוטומטיים מדורגים לשינויים בהגדרות.
• שדרוגים אוטומטיים מתגלגלים ללא השפעה על הזמינות של Kafka.
• אם מתרחש כשל, CFK משחזר פוד של Kafka עם אותו מזהה ברוקר של Kafka, אותה הגדרה ואותם נפחי אחסון מתמיד.
• מודעות אוטומטית למתלים כדי לפזר רפליקות של מחיצה בין מתלים (או אזורים) שונים, לשפר את הזמינות של ברוקרי Kafka ולהגביל את הסיכון לאובדן נתונים.
• תמיכה בייצוא של מדדים מצטברים ל-Prometheus.

ארכיטקטורת פריסה

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

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

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

זמינות ותוכנית התאוששות מאסון (DR)

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

אשכולות Kubernetes עם זמינות גבוהה ב- Google Cloud מסתמכים על אשכולות אזוריים שמשתרעים על כמה צמתים ואזורי זמינות. ההגדרה הזו משפרת את עמידות המערכת בפני תקלות, את יכולת ההתאמה שלה ואת יתירות המיקום הגיאוגרפי. ההגדרה הזו מאפשרת גם לבצע עדכונים ותחזוקה בזמן שהמערכת פועלת, וגם מספקת הסכמי רמת שירות (SLA) לגבי זמן פעולה וזמינות. מידע נוסף זמין במאמר אשכולות אזוריים.

תרשים פריסה

הדיאגרמה הבאה מציגה אשכול Kafka שפועל בכמה צמתים ואזורים באשכול GKE:

בתרשים, ה-StatefulSet של Kafka נפרס על פני שלושה צמתים בשלושה אזורים שונים. אפשר לשלוט בהגדרה הזו באמצעות הגדרת הכללים affinity ו-topology spread של Pod required במפרט של המשאב המותאם אישית Kafka.

אם אזור אחד נכשל, באמצעות ההגדרה המומלצת, ‏ GKE מתזמן מחדש את ה-Pods בצמתים חדשים ומשכפל את הנתונים מהרפליקות שנותרו, גם עבור Kafka וגם עבור Zookeeper.

בתרשים הבא מוצג ZooKeeper StatefulSet שנפרס בשלושה צמתים בשלושה אזורים שונים:

עלויות

במסמך הזה משתמשים ברכיבים הבאים של Google Cloud, והשימוש בהם כרוך בתשלום:

• GKE
• Managed Service for Prometheus
• Compute Engine

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

משתמשים חדשים של Google Cloud ? יכול להיות שאתם זכאים לתקופת ניסיון בחינם.

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

לפני שמתחילים

נכנסים לחשבון Google Cloud . אם אתם משתמשים חדשים ב- Google Cloud, צרו חשבון כדי שתוכלו להעריך את הביצועים של המוצרים שלנו בתרחישים מהעולם האמיתי. לקוחות חדשים מקבלים בחינם גם קרדיט בשווי 300$ להרצה, לבדיקה ולפריסה של עומסי העבודה.

התקינו את ה-CLI של Google Cloud.

הערה: אם התקנתם את ה-CLI של gcloud, השתמשו בפקודה gcloud components update כדי לבדוק אם הגרסה העדכנית מותקנת.

אם אתם משתמשים בספק זהויות חיצוני (IdP), קודם אתם צריכים להיכנס ל-CLI של gcloud באמצעות המאגר המאוחד לניהול זהויות.

כדי לאתחל את ה-CLI של gcloud, הריצו את הפקודה הבאה:

gcloud init

יוצרים או בוחרים Google Cloud פרויקט.

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

• Select a project: כדי לבחור פרויקט לא צריך תפקיד IAM ספציפי – אפשר לבחור כל פרויקט שקיבלתם בו תפקיד.
• יצירת פרויקט: כדי ליצור פרויקט, צריך את התפקיד Project Creator (יצירת פרויקטים) (roles/resourcemanager.projectCreator), שכולל את ההרשאה resourcemanager.projects.create. איך מקצים תפקידים

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

• יוצרים Google Cloud פרויקט:

  gcloud projects create PROJECT_ID

  מחליפים את PROJECT_ID בשם של פרויקט Google Cloud שיוצרים.

• בוחרים את הפרויקט שיצרתם: Google Cloud

  gcloud config set project PROJECT_ID

  מחליפים את PROJECT_ID בשם הפרויקט ב- Google Cloud .

מוודאים שהחיוב מופעל בפרויקט Google Cloud .

מפעילים את ממשקי ה-API של GKE,‏ Backup for GKE,‏ Compute Engine,‏ Identity and Access Management ומנהל המשאבים:

תפקידים שנדרשים להפעלת ממשקי API

כדי להפעיל ממשקי API, צריך את תפקיד ה-IAM 'אדמין של Service Usage' (roles/serviceusage.serviceUsageAdmin), שכולל את ההרשאה serviceusage.services.enable. איך מקצים תפקידים

gcloud services enable compute.googleapis.com iam.googleapis.com container.googleapis.com gkebackup.googleapis.com cloudresourcemanager.googleapis.com

התקינו את ה-CLI של Google Cloud.

הערה: אם התקנתם את ה-CLI של gcloud, השתמשו בפקודה gcloud components update כדי לבדוק אם הגרסה העדכנית מותקנת.

אם אתם משתמשים בספק זהויות חיצוני (IdP), קודם אתם צריכים להיכנס ל-CLI של gcloud באמצעות המאגר המאוחד לניהול זהויות.

כדי לאתחל את ה-CLI של gcloud, הריצו את הפקודה הבאה:

gcloud init

יוצרים או בוחרים Google Cloud פרויקט.

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

• Select a project: כדי לבחור פרויקט לא צריך תפקיד IAM ספציפי – אפשר לבחור כל פרויקט שקיבלתם בו תפקיד.
• יצירת פרויקט: כדי ליצור פרויקט, צריך את התפקיד Project Creator (יצירת פרויקטים) (roles/resourcemanager.projectCreator), שכולל את ההרשאה resourcemanager.projects.create. איך מקצים תפקידים

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

• יוצרים Google Cloud פרויקט:

  gcloud projects create PROJECT_ID

  מחליפים את PROJECT_ID בשם של פרויקט Google Cloud שיוצרים.

• בוחרים את הפרויקט שיצרתם: Google Cloud

  gcloud config set project PROJECT_ID

  מחליפים את PROJECT_ID בשם הפרויקט ב- Google Cloud .

מוודאים שהחיוב מופעל בפרויקט Google Cloud .

מפעילים את ממשקי ה-API של GKE,‏ Backup for GKE,‏ Compute Engine,‏ Identity and Access Management ומנהל המשאבים:

תפקידים שנדרשים להפעלת ממשקי API

כדי להפעיל ממשקי API, צריך את תפקיד ה-IAM 'אדמין של Service Usage' (roles/serviceusage.serviceUsageAdmin), שכולל את ההרשאה serviceusage.services.enable. איך מקצים תפקידים

gcloud services enable compute.googleapis.com iam.googleapis.com container.googleapis.com gkebackup.googleapis.com cloudresourcemanager.googleapis.com

1. מעניקים תפקידים לחשבון המשתמש. מריצים את הפקודה הבאה לכל אחד מהתפקידים הבאים ב-IAM: role/storage.objectViewer, role/logging.logWriter, roles/container.clusterAdmin, role/container.serviceAgent, roles/iam.serviceAccountAdmin, roles/serviceusage.serviceUsageAdmin, roles/iam.serviceAccountAdmin

   gcloud projects add-iam-policy-binding PROJECT_ID --member="user:USER_IDENTIFIER" --role=ROLE

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

   • ‫PROJECT_ID: מזהה הפרויקט.
   • ‫USER_IDENTIFIER: המזהה של חשבון המשתמש . לדוגמה, [e-mail].
   • ‫ROLE: תפקיד ה-IAM שאתם מקצים לחשבון המשתמש.

הכנת הסביבה

במדריך הזה תשתמשו ב-Cloud Shell כדי לנהל משאבים שמתארחים ב- Google Cloud. ב-Cloud Shell מותקן מראש התוכנה שדרושה למדריך הזה, כולל kubectl,‏ ה-CLI של gcloud,‏ Helm ו-Terraform.

כדי להגדיר את הסביבה באמצעות Cloud Shell:

1. מפעילים סשן של Cloud Shell מGoogle Cloud המסוף על ידי לחיצה על Activate Cloud Shell . Google Cloud ייפתח סשן בחלונית התחתונה של מסוף Google Cloud .

2. הגדרת משתני סביבה:

   export PROJECT_ID=PROJECT_ID
   export KUBERNETES_CLUSTER_PREFIX=kafka
   export REGION=us-central1
   

   מחליפים את PROJECT_ID: your Google Cloud במזהה הפרויקט.

3. משכפלים את המאגר ב-GitHub:

   git clone https://github.com/GoogleCloudPlatform/kubernetes-engine-samples
   

4. עוברים לספריית העבודה:

   cd kubernetes-engine-samples/streaming
   

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

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

אפשר להתקין את האופרטור באמצעות אשכול Standard או Autopilot.

רגילה

בתרשים הבא מוצג אשכול GKE פרטי אזורי מסוג Standard שנפרס בשלושה אזורים שונים:

כדי לפרוס את התשתית הזו, מריצים את הפקודות הבאות מ-Cloud Shell:

export GOOGLE_OAUTH_ACCESS_TOKEN=$(gcloud auth print-access-token)
terraform -chdir=kafka/terraform/gke-standard init
terraform -chdir=kafka/terraform/gke-standard apply -var project_id=${PROJECT_ID} \
  -var region=${REGION} \
  -var cluster_prefix=${KUBERNETES_CLUSTER_PREFIX}

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

‫Terraform יוצר את המשאבים הבאים:

• רשת VPC ותת-רשת פרטית לצמתים של Kubernetes.
• נתב לגישה לאינטרנט דרך NAT.
• אשכול GKE פרטי באזור us-central1.
• ‫2 מאגרי צמתים עם התאמה אוטומטית לעומס (1-2 צמתים לכל אזור, צומת אחד לכל אזור לכל הפחות)
• ServiceAccount עם הרשאות רישום ביומן ומעקב.
• גיבוי ל-GKE לצורך התאוששות מאסון.
• השירות המנוהל של Google Cloud ל-Prometheus לניטור אשכולות.

הפלט אמור להיראות כך:

...
Apply complete! Resources: 14 added, 0 changed, 0 destroyed.

Outputs:

kubectl_connection_command = "gcloud container clusters get-credentials kafka-cluster --region us-central1"

טייס אוטומטי

התרשים הבא מציג אשכול פרטי אזורי של Autopilot GKE:

כדי לפרוס את התשתית, מריצים את הפקודות הבאות מ-Cloud Shell:

export GOOGLE_OAUTH_ACCESS_TOKEN=$(gcloud auth print-access-token)
terraform -chdir=kafka/terraform/gke-autopilot init
terraform -chdir=kafka/terraform/gke-autopilot apply -var project_id=${PROJECT_ID} \
  -var region=${REGION} \
  -var cluster_prefix=${KUBERNETES_CLUSTER_PREFIX}

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

‫Terraform יוצר את המשאבים הבאים:

• רשת VPC ותת-רשת פרטית לצמתים של Kubernetes.
• נתב לגישה לאינטרנט דרך NAT.
• אשכול GKE פרטי באזור us-central1.
• ServiceAccount עם הרשאות רישום ביומן ומעקב
• השירות המנוהל של Google Cloud ל-Prometheus לניטור אשכולות.

הפלט אמור להיראות כך:

...
Apply complete! Resources: 12 added, 0 changed, 0 destroyed.

Outputs:

kubectl_connection_command = "gcloud container clusters get-credentials kafka-cluster --region us-central1"

התחברות לאשכול

מגדירים את kubectl לתקשורת עם האשכול:

gcloud container clusters get-credentials ${KUBERNETES_CLUSTER_PREFIX}-cluster --region ${REGION}

פריסת אופרטור CFK באשכול

בקטע הזה פורסים את האופרטור Confluent for Kubernetes ‏ (CFK) באמצעות תרשים Helm, ואז פורסים אשכול Kafka.

1. מוסיפים את מאגר התרשימים של Confluent Helm:

   helm repo add confluentinc https://packages.confluent.io/helm
   

2. מוסיפים מרחב שמות לאופרטור CFK ולמקלט Kafka:

   kubectl create ns kafka
   

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

   helm install confluent-operator confluentinc/confluent-for-kubernetes -n kafka
   

   כדי לאפשר ל-CFK לנהל משאבים בכל מרחבי השמות, מוסיפים את הפרמטר --set-namespaced=false לפקודת Helm.

4. מוודאים שה-Operator של Confluent נפרס בהצלחה באמצעות Helm:

   helm ls -n kafka
   

   הפלט אמור להיראות כך:

   NAME                  NAMESPACE  REVISION UPDATED                                  STATUS      CHART                                APP VERSION
   confluent-operator    kafka      1        2023-07-07 10:57:45.409158 +0200 CEST    deployed    confluent-for-kubernetes-0.771.13    2.6.0
   

פריסת Kafka

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

הגדרה בסיסית

ההגדרה הבסיסית של מופע Kafka כוללת את הרכיבים הבאים:

• שלוש רפליקות של ברוקרים של Kafka, עם מינימום של שתי רפליקות זמינות שנדרשות לעקביות של האשכול.
• שלוש רפליקות של צמתי ZooKeeper, שיוצרות אשכול.
• שני מאזינים של Kafka: אחד ללא אימות, ואחד שמשתמש באימות TLS עם אישור שנוצר על ידי CFK.
• ההגדרות Java MaxHeapSize ו-MinHeapSize מוגדרות ל-4 GB עבור Kafka.
• הקצאת משאבי CPU של בקשת CPU אחת ו-2 מגבלות CPU, וכן בקשות ומגבלות של זיכרון בנפח 5GB ל-Kafka (4GB לשירות הראשי ו-0.5GB לייצוא המדדים) ו-3GB ל-Zookeeper (2GB לשירות הראשי ו-0.5GB לייצוא המדדים).
• נפח אחסון של 100GB מוקצה לכל Pod באמצעות premium-rwo storageClass, ‏ 100 לנתוני Kafka ו-90/10 לנתוני Zookeeper/יומן.
• הגדרת Tolerations,‏ nodeAffinities ו-podAntiAffinities לכל עומס עבודה, כדי להבטיח חלוקה נכונה בין הצמתים, תוך שימוש במאגרי הצמתים המתאימים ובאזורים שונים.
• התקשורת בתוך האשכול מאובטחת באמצעות אישורים עם חתימה עצמית, באמצעות רשות אישורים שאתם מספקים.

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

יצירת אשכול Kafka בסיסי

1. יצירת צמד CA:

   openssl genrsa -out ca-key.pem 2048
   openssl req -new -key ca-key.pem -x509 \
     -days 1000 \
     -out ca.pem \
     -subj "/C=US/ST=CA/L=Confluent/O=Confluent/OU=Operator/CN=MyCA"
   

   ‫Confluent for Kubernetes מספק אישורים שנוצרים באופן אוטומטי לרכיבי Confluent Platform לשימוש בהצפנת רשת TLS. צריך ליצור ולספק רשות אישורים (CA).

2. יוצרים סוד ב-Kubernetes עבור רשות האישורים:

   kubectl create secret tls ca-pair-sslcerts --cert=ca.pem --key=ca-key.pem -n kafka
   

   השם של הסוד הוא predefined

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

   kubectl apply -n kafka -f kafka-confluent/manifests/01-basic-cluster/my-cluster.yaml
   

   הפקודה הזו יוצרת משאב מותאם אישית של Kafka ומשאב מותאם אישית של Zookeeper של אופרטור CFK, שכוללים בקשות ומגבלות של CPU וזיכרון, בקשות לאחסון בלוקים, ו-taints ו-affinities כדי להפיץ את ה-Pods שהוקצו על פני צמתי Kubernetes.

4. ממתינים כמה דקות בזמן ש-Kubernetes מפעיל את עומסי העבודה הנדרשים:

   kubectl wait pods -l app=my-cluster --for condition=Ready --timeout=300s -n kafka
   

5. מוודאים שעומסי העבודה של Kafka נוצרו:

   kubectl get pod,svc,statefulset,deploy,pdb -n kafka
   

   הפלט אמור להיראות כך:

   NAME                                    READY   STATUS  RESTARTS   AGE
   pod/confluent-operator-864c74d4b4-fvpxs   1/1   Running   0        49m
   pod/my-cluster-0                        1/1   Running   0        17m
   pod/my-cluster-1                        1/1   Running   0        17m
   pod/my-cluster-2                        1/1   Running   0        17m
   pod/zookeeper-0                         1/1   Running   0        18m
   pod/zookeeper-1                         1/1   Running   0        18m
   pod/zookeeper-2                         1/1   Running   0        18m
   
   NAME                          TYPE      CLUSTER-IP   EXTERNAL-IP   PORT(S)                                                        AGE
   service/confluent-operator    ClusterIP   10.52.13.164   <none>      7778/TCP                                                       49m
   service/my-cluster            ClusterIP   None         <none>      9092/TCP,8090/TCP,9071/TCP,7203/TCP,7777/TCP,7778/TCP,9072/TCP   17m
   service/my-cluster-0-internal   ClusterIP   10.52.2.242  <none>      9092/TCP,8090/TCP,9071/TCP,7203/TCP,7777/TCP,7778/TCP,9072/TCP   17m
   service/my-cluster-1-internal   ClusterIP   10.52.7.98   <none>      9092/TCP,8090/TCP,9071/TCP,7203/TCP,7777/TCP,7778/TCP,9072/TCP   17m
   service/my-cluster-2-internal   ClusterIP   10.52.4.226  <none>      9092/TCP,8090/TCP,9071/TCP,7203/TCP,7777/TCP,7778/TCP,9072/TCP   17m
   service/zookeeper             ClusterIP   None         <none>      2181/TCP,7203/TCP,7777/TCP,3888/TCP,2888/TCP,7778/TCP          18m
   service/zookeeper-0-internal  ClusterIP   10.52.8.52   <none>      2181/TCP,7203/TCP,7777/TCP,3888/TCP,2888/TCP,7778/TCP          18m
   service/zookeeper-1-internal  ClusterIP   10.52.12.44  <none>      2181/TCP,7203/TCP,7777/TCP,3888/TCP,2888/TCP,7778/TCP          18m
   service/zookeeper-2-internal  ClusterIP   10.52.12.134   <none>      2181/TCP,7203/TCP,7777/TCP,3888/TCP,2888/TCP,7778/TCP          18m
   
   NAME                        READY   AGE
   statefulset.apps/my-cluster   3/3   17m
   statefulset.apps/zookeeper  3/3   18m
   
   NAME                               READY   UP-TO-DATE   AVAILABLE   AGE
   deployment.apps/confluent-operator   1/1   1          1         49m
   
   NAME                                  MIN AVAILABLE   MAX UNAVAILABLE   ALLOWED DISRUPTIONS   AGE
   poddisruptionbudget.policy/my-cluster   N/A           1               1                   17m
   poddisruptionbudget.policy/zookeeper  N/A           1               1                   18m
   

המפעיל יוצר את המשאבים הבאים:

• שני StatefulSets ל-Kafka ול-ZooKeeper.
• שלושה פודים לשכפול של ברוקר Kafka.
• שלושה רכיבי Pod לשכפול של ZooKeeper.
• שני משאבי PodDisruptionBudget, כדי להבטיח שיהיה לכל היותר עותק אחד לא זמין לשמירה על עקביות האשכול.
• השירות my-cluster שמשמש כשרת bootstrap ללקוחות Kafka שמתחברים מתוך אשכול Kubernetes. כל מאזיני Kafka הפנימיים זמינים בשירות הזה.
• השירות zookeeper שמאפשר לשרתי Kafka להתחבר לצמתי ZooKeeper כלקוחות.

אימות וניהול משתמשים

בקטע הזה מוסבר איך להפעיל את האימות וההרשאה כדי לאבטח את רכיבי ה-Kafka Listeners ולשתף את פרטי הכניסה עם לקוחות.

‫Confluent for Kubernetes תומך בשיטות אימות שונות ל-Kafka, כמו:

• אימות SASL/PLAIN: הלקוחות משתמשים בשם משתמש ובסיסמה לאימות. שם המשתמש והסיסמה מאוחסנים בצד השרת בסוד של Kubernetes.
• ‫SASL/PLAIN עם אימות LDAP: הלקוחות משתמשים בשם משתמש ובסיסמה לצורך אימות. פרטי הכניסה מאוחסנים בשרת LDAP.
• אימות mTLS: לקוחות משתמשים באישור TLS לצורך אימות.

מגבלות

• ‫CFK לא מספק משאבים בהתאמה אישית לניהול משתמשים. עם זאת, אפשר לאחסן פרטי כניסה ב-Secrets ולהפנות ל-Secrets במפרטים של מאזינים.
• למרות שאין Custom Resource לניהול רשימות ACL באופן ישיר, ב-Confluent for Kubernetes הרשמי יש הנחיות להגדרת רשימות ACL באמצעות Kafka CLI.

יצירת משתמש

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

• קלאסטר Kafka עם אימות מבוסס-סיסמה (SASL/PLAIN) שמופעל באחד מהמאזינים
• ‫KafkaTopic עם 3 העתקים
• פרטי כניסה של משתמשים עם הרשאות קריאה וכתיבה

1. יצירת סוד עם פרטי כניסה של משתמש:

   export USERNAME=my-user
   export PASSWORD=$(openssl rand -base64 12)
   kubectl create secret generic my-user-credentials -n kafka \
     --from-literal=plain-users.json="{\"$USERNAME\":\"$PASSWORD\"}"
   

   פרטי הכניסה צריכים להיות מאוחסנים בפורמט הבא:

   {
   "username1": "password1",
   "username2": "password2",
   ...
   "usernameN": "passwordN"
   }
   

2. הגדרת אשכול Kafka לשימוש במאזין עם אימות מבוסס-סיסמה SCRAM-SHA-512 ביציאה 9094:

   kubectl apply -n kafka -f kafka-confluent/manifests/02-auth/my-cluster.yaml
   

3. מגדירים נושא ו-Pod של לקוח כדי ליצור אינטראקציה עם אשכול Kafka ולהריץ פקודות Kafka:

   kubectl apply -n kafka -f kafka-confluent/manifests/02-auth/my-topic.yaml
   kubectl apply -n kafka -f kafka-confluent/manifests/02-auth/kafkacat.yaml
   

   ‫GKE מעלה את ה-Secret‏ my-user-credentials ל-Pod הלקוח כVolume.

4. כשה-Pod של הלקוח מוכן, מתחברים אליו ומתחילים ליצור ולצרוך הודעות באמצעות פרטי הכניסה שסופקו:

   kubectl wait pod kafkacat --for=condition=Ready --timeout=300s -n kafka
   kubectl exec -it kafkacat -n kafka -- /bin/sh
   

5. יוצרים הודעה באמצעות פרטי הכניסה של my-user ואז צורכים את ההודעה כדי לאמת את הקבלה שלה.

   export USERNAME=$(cat /my-user/plain-users.json|cut -d'"' -f 2)
   export PASSWORD=$(cat /my-user/plain-users.json|cut -d'"' -f 4)
   echo "Message from my-user" |kcat \
     -b my-cluster.kafka.svc.cluster.local:9094 \
     -X security.protocol=SASL_SSL \
     -X sasl.mechanisms=PLAIN \
     -X sasl.username=$USERNAME \
     -X sasl.password=$PASSWORD  \
     -t my-topic -P
   kcat -b my-cluster.kafka.svc.cluster.local:9094 \
     -X security.protocol=SASL_SSL \
     -X sasl.mechanisms=PLAIN \
     -X sasl.username=$USERNAME \
     -X sasl.password=$PASSWORD  \
     -t my-topic -C
   

   הפלט אמור להיראות כך:

   Message from my-user
   % Reached end of topic my-topic [1] at offset 1
   % Reached end of topic my-topic [2] at offset 0
   % Reached end of topic my-topic [0] at offset 0
   

   מקלידים CTRL+C כדי להפסיק את התהליך של הצרכן. אם מופיעה Connect refused שגיאה, מחכים כמה דקות ומנסים שוב.

6. יציאה מהמעטפת של ה-Pod

   exit
   

גיבויים ותוכנית התאוששות מאסון (DR)

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

אפשר להשתמש בגיבוי ל-GKE כדי לגבות:

• מניפסטים של משאבים ב-Kubernetes.
• משאבים מותאמים אישית של Confluent API וההגדרות שלהם, שחולצו משרת ה-API של Kubernetes באשכול שעובר גיבוי.
• נפחי אחסון שתואמים למשאבי PersistentVolumeClaim שנמצאו במניפסטים.

אפשר גם לבצע גיבוי ידני של אשכול Kafka. אתם צריכים לגבות:

• ההגדרה של Kafka, שכוללת את כל המשאבים המותאמים אישית של Confluent API, כמו KafkaTopics או Connect
• הנתונים, שמאוחסנים ב-PersistentVolumes של ברוקרי Kafka

אחסון של מניפסטים של משאבי Kubernetes, כולל הגדרות Confluent, במאגרי Git יכול לבטל את הצורך בגיבוי נפרד של הגדרות Kafka, כי אפשר להחיל מחדש את המשאבים על אשכול Kubernetes חדש כשצריך.

כדי להגן על שחזור נתונים של Kafka בתרחישים שבהם אבד מופע של שרת Kafka או אשכול Kubernetes שבו Kafka נפרס, מומלץ להגדיר את סוג האחסון (storage class) של Kubernetes שמשמשת להקצאת נפחים עבור ברוקרים של Kafka עם האפשרות reclaimPolicy שמוגדרת ל-Retain. מומלץ גם לצלם תמונות מצב של נפחי Kafka broker.

המאניפסט הבא מתאר StorageClass שמשתמש באפשרות reclaimPolicyRetain:

apiVersion: storage.k8s.io/v1
kind: StorageClass
metadata:
  name: premium-rwo-retain
...
reclaimPolicy: Retain
volumeBindingMode: WaitForFirstConsumer

בדוגמה הבאה מוצג StorageClass שנוסף ל-spec של משאב מותאם אישית של אשכול Kafka:

...
spec:
  ...
  dataVolumeCapacity: 100Gi
  storageClass:
  name: premium-rwo-retain

במקרה כזה, כרכים קבועים (PersistentVolumes) שהוקצו באמצעות סוג האחסון (storage class) לא נמחקים גם כשמחקתם את התביעה המתאימה לכרך קבוע (PersistentVolumeClaim).

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

1. החלת משאבים מותאמים אישית קיימים של Confluent ‏ (Kafka,‏ KafkaTopic,‏ Zookeeper וכו') על אשכול Kubernetes חדש
2. מעדכנים את ה-PersistentVolumeClaims עם השם של מופעי ברוקר Kafka החדשים ל-PersistentVolumes הישנים באמצעות המאפיין spec.volumeName ב-PersistentVolumeClaim.

הסרת המשאבים

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

מחיקת הפרויקט

זהירות: כשמוחקים פרויקט:

• התוכן של הפרויקט נמחק כולו. אם השתמשתם בפרויקט קיים כדי לתרגל את המשימות במדריך הזה, כשתמחקו אותו תימחק גם כל העבודה שביצעתם באותו פרויקט.
• מאבדים את מזהה הפרויקט בהתאמה אישית. כשיצרתם את הפרויקט, יכולתם לבחור לו מזהה פרויקט בהתאמה אישית כדי להשתמש בו בעתיד. כדי לשמור על כתובות ה-URL שמשתמשות במזהה הפרויקט, כמו appspot.com, צריך למחוק את המשאבים שנבחרו בפרויקט אבל לא את הפרויקט כולו.

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

כדי למחוק Google Cloud פרויקט:

gcloud projects delete PROJECT_ID

מחיקת המשאבים הבודדים

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

1. הגדרת משתני סביבה:

   export PROJECT_ID=PROJECT_ID
   export KUBERNETES_CLUSTER_PREFIX=kafka
   export REGION=us-central1
   

2. מריצים את הפקודה terraform destroy:

   export GOOGLE_OAUTH_ACCESS_TOKEN=$(gcloud auth print-access-token)
   terraform -chdir=kafka/terraform/FOLDER destroy -var project_id=${PROJECT_ID}   \
     -var region=${REGION}  \
     -var cluster_prefix=${KUBERNETES_CLUSTER_PREFIX}
   

   מחליפים את FOLDER ב-gke-autopilot או ב-gke-standard.

   כשמופיעה בקשה, כותבים yes.

3. חיפוש כל הדיסקים שלא צורפו:

   export disk_list=$(gcloud compute disks list --filter="-users:* AND labels.name=${KUBERNETES_CLUSTER_PREFIX}-cluster" --format "value[separator=|](name,zone)")
   

4. מוחקים את הדיסקים:

   for i in $disk_list; do
     disk_name=$(echo $i| cut -d'|' -f1)
     disk_zone=$(echo $i| cut -d'|' -f2|sed 's|.*/||')
     echo "Deleting $disk_name"
     gcloud compute disks delete $disk_name --zone $disk_zone --quiet
   done
   

המאמרים הבאים

• כדאי להעמיק את הקריאה ולהכיר דוגמאות לארכיטקטורות, תרשימים ושיטות מומלצות בנושאי Google Cloud. כל אלה זמינים במרכז הארכיטקטורה של Cloud.
• איך מגבים את האשכולות באמצעות Backup for GKE