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

במדריך מוסבר איך להשתמש באופרטור Strimzi כדי לפרוס אשכולות של Apache Kafka.

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

אופרטורים הם תוספי תוכנה שמשתמשים במשאבים מותאמים אישית כדי לנהל אפליקציות ורכיבים שלהן. במאמר תבנית אופרטור במסמכי הקוד הפתוח של Kubernetes אפשר לקרוא מידע נוסף על הסיבות לשימוש באופרטורים. האופרטור Strimzi מציע גמישות באפשרויות הפריסה ומאפשר להשתמש ב-taints וב-tolerances של Kubernetes כדי להריץ את Kafka בצמתים ייעודיים.

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

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

מטרות

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

יתרונות

היתרונות של Strimzi כוללים:

  • אופרטורים של Strimzi מספקים גישה פשוטה ומקורית ל-Kubernetes לניהול אשכולות Kafka. ב-Strimzi משתמשים במשאבים בהתאמה אישית שמייצגים נושאים ומשתמשים ב-Kafka, וכך ניהול האשכולות פשוט יותר ותואם לשיטות המומלצות של Kubernetes.
  • Strimzi מתעדף אבטחה כברירת מחדל על ידי יצירת אישורים למאזינים ותמיכה בשיטות אימות מאובטחות כמו TLS,‏ SCRAM-SHA ו-OAuth. ‫Strimzi מטפל גם ב-NetworkPolicies לכל מאזיני Kafka.
  • ל-Strimzi אין תלות בגורמים חיצוניים. הוא כולל אשכולות של Kafka ושל ZooKeeper עם כלי מובנה לייצוא מדדים, כך שלא צריך להשתמש בכלים נוספים. אפשר גם לשנות את ההגדרות של הברוקר כדי לעמוד בדרישות ספציפיות.

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

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

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

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

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

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

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

שימוש בכמה צמתים ואזורים הוא חיוני כדי להשיג אשכול Kubernetes עם זמינות גבוהה ב- Google Cloud מהסיבות הבאות:

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

תרשים פריסה

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

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

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

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

המשאב המותאם אישית StrimziPodSet

במדריך הזה נעשה שימוש במשאב המותאם אישית StrimziPodSet שהוצג בגרסה 0.29 של Strimzi, במקום ב-StatefulSets.

המשאבים StrimziPodSet מאפשרים להגדיל את יכולת ההתאמה של האשכול, ומאפשרים להעביר אפשרויות הגדרה כדי לבצע שינויים ב-Pods ברמת פירוט גבוהה יותר. המשאב StrimziPodSet מופעל כברירת מחדל ב-Strimzi מגרסה 0.35 ואילך.

עלויות

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

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

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

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

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

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

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

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

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

    gcloud init

  5. יוצרים או בוחרים פרויקט ב- 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 .

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

  7. מפעילים את ממשקי ה-API של Compute Engine,‏ IAM,‏ GKE,‏ Backup for GKE ו-Resource Manager, אם הם עדיין לא מופעלים:

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

    כדי להפעיל ממשקי API, צריך את ההרשאה serviceusage.services.enable. אם יצרתם את הפרויקט, סביר להניח שכבר יש לכם את ההרשאה הזו דרך התפקיד 'בעלים' (roles/owner). אחרת, תוכלו לקבל את ההרשאה הזו דרך התפקיד 'אדמין בממשק Service Usage' (roles/serviceusage.serviceUsageAdmin). איך מקצים תפקידים 

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

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

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

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

    gcloud init

  11. יוצרים או בוחרים פרויקט ב- 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 .

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

  13. מפעילים את ממשקי ה-API של Compute Engine,‏ IAM,‏ GKE,‏ Backup for GKE ו-Resource Manager, אם הם עדיין לא מופעלים:

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

    כדי להפעיל ממשקי API, צריך את ההרשאה serviceusage.services.enable. אם יצרתם את הפרויקט, סביר להניח שכבר יש לכם את ההרשאה הזו דרך התפקיד 'בעלים' (roles/owner). אחרת, תוכלו לקבל את ההרשאה הזו דרך התפקיד 'אדמין בממשק Service Usage' (roles/serviceusage.serviceUsageAdmin). איך מקצים תפקידים 

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

  14. מעניקים תפקידים לחשבון המשתמש. מריצים את הפקודה הבאה לכל אחד מהתפקידים הבאים ב-IAM: roles/storage.objectViewer, roles/logging.logWriter, roles/container.clusterAdmin, roles/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: המזהה של חשבון המשתמש . לדוגמה, myemail@example.com.
    • ‫ROLE: תפקיד ה-IAM שאתם מקצים לחשבון המשתמש.

הכנת הסביבה

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

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

  1. מפעילים סשן של Cloud Shell במסוף Google Cloud : לוחצים על סמל ההפעלה של Cloud Shell 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 אזורי עם זמינות גבוהה וצמתים פרטיים. כדי לאפשר גישה ציבורית למישור הבקרה: כדי להגביל את הגישה, מגדירים את האשכול כך שמישור הבקרה ייחשף רק בנקודת הקצה הפנימית.

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

רגילה

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

כדי לפרוס את התשתית הזו, מריצים את הפקודות הבאות מ-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 strimzi-cluster --region us-central1"

טייס אוטומטי

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

כדי לפרוס את התשתית, מריצים את הפקודות הבאות מ-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 strimzi-cluster --region us-central1"

חיבור לאוסף

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

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

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

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

  1. מוסיפים את מאגר Strimzi Helm Chart:

    helm repo add strimzi https://strimzi.io/charts/

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

    kubectl create ns kafka

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

    helm install strimzi-operator strimzi/strimzi-kafka-operator -n kafka

    כדי לפרוס את Strimzi Cluster Operator ואת אשכולי Kafka במרחבי שמות שונים, מוסיפים את הפרמטר --set watchNamespaces="{kafka-namespace,kafka-namespace-2,...}" לפקודת helm.

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

    helm ls -n kafka

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

    NAME            NAMESPACE    REVISION    UPDATED                              STATUS    CHART                        APP VERSION
strimzi-operator    kafka      1       2023-06-27 11:22:15.850545 +0200 CEST    deployed    strimzi-kafka-operator-0.35.0    0.35.0

פריסת Kafka

אחרי שהאופרטור נפרס באשכול, אפשר לפרוס מופע של אשכול Kafka.

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

הגדרה בסיסית

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

  • שלוש רפליקות של ברוקרים של Kafka, עם מינימום של שתי רפליקות זמינות שנדרשות לעקביות של האשכול.
  • שלוש רפליקות של צומתי ZooKeeper, שיוצרות אשכול.
  • שני מאזינים של Kafka: אחד בלי אימות, ואחד שמשתמש באימות TLS עם אישור שנוצר על ידי Strimzi.
  • הערכים של Java MaxHeapSize ו-MinHeapSize מוגדרים ל-4 GB עבור Kafka ול-2 GB עבור ZooKeeper.
  • הקצאת משאבי CPU של בקשת CPU אחת ו-2 מגבלות CPU גם ל-Kafka וגם ל-ZooKeeper, יחד עם בקשות ומגבלות של זיכרון בנפח 5GB ל-Kafka (4GB לשירות הראשי ו-0.5GB ל-metrics exporter) ו-2.5GB ל-ZooKeeper (2GB לשירות הראשי ו-0.5GB ל-metrics exporter).
  • ‫Entity-operator עם הבקשות והמגבלות הבאות:
    • ‫tlsSidecar: 100 m/500 m CPU ו-128 Mi זיכרון.
    • ‫topicOperator: מעבד (CPU) של ‎100 m/500 m וזיכרון של ‎512 Mi.
    • ‫userOperator: 500 m CPU ו-2 Gi זיכרון.
  • נפח אחסון של 100GB‎ מוקצה לכל Pod באמצעות premium-rwo storageClass.
  • הגדרות של Tolerations,‏ nodeAffinities ו-podAntiAffinities לכל עומס עבודה, כדי להבטיח חלוקה נכונה בין הצמתים, תוך שימוש במאגרי הצמתים המתאימים ובאזורים שונים.
  • התקשורת בתוך האשכול מאובטחת באמצעות אישורים בחתימה עצמית: רשויות אישורים (CA) נפרדות לאשכול וללקוחות (mTLS). אפשר גם להגדיר שימוש ברשות אישורים אחרת.

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

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

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

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

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

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

    kubectl wait kafka/my-cluster --for=condition=Ready --timeout=600s -n kafka

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

    kubectl get pod,service,deploy,pdb -l=strimzi.io/cluster=my-cluster -n kafka

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

    NAME                                            READY   STATUS  RESTARTS   AGE
pod/my-cluster-entity-operator-848698874f-j5m7f   3/3   Running   0        44m
pod/my-cluster-kafka-0                          1/1   Running   0        5m
pod/my-cluster-kafka-1                          1/1   Running   0        5m
pod/my-cluster-kafka-2                          1/1   Running   0        5m
pod/my-cluster-zookeeper-0                      1/1   Running   0        6m
pod/my-cluster-zookeeper-1                      1/1   Running   0        6m
pod/my-cluster-zookeeper-2                      1/1   Running   0        6m

NAME                                TYPE      CLUSTER-IP   EXTERNAL-IP   PORT(S)                             AGE
service/my-cluster-kafka-bootstrap  ClusterIP   10.52.8.80   <none>      9091/TCP,9092/TCP,9093/TCP          5m
service/my-cluster-kafka-brokers    ClusterIP   None         <none>      9090/TCP,9091/TCP,9092/TCP,9093/TCP   5m
service/my-cluster-zookeeper-client   ClusterIP   10.52.11.144   <none>      2181/TCP                            6m
service/my-cluster-zookeeper-nodes  ClusterIP   None         <none>      2181/TCP,2888/TCP,3888/TCP          6m

NAME                                       READY   UP-TO-DATE   AVAILABLE   AGE
deployment.apps/my-cluster-entity-operator   1/1   1          1         44m

NAME                                            MIN AVAILABLE   MAX UNAVAILABLE   ALLOWED DISRUPTIONS   AGE
poddisruptionbudget.policy/my-cluster-kafka     2             N/A             1                   5m
poddisruptionbudget.policy/my-cluster-zookeeper   2             N/A             1                   6m

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

  • שני StrimziPodSets ל-Kafka ול-ZooKeeper.
  • שלושה פודים לשכפול של ברוקר Kafka.
  • שלושה פודים לרפליקות של ZooKeeper.
  • ‫2 PodDisruptionBudgets, כדי להבטיח זמינות מינימלית של שני עותקים לשמירה על עקביות האשכול.
  • שירות בשם my-cluster-kafka-bootstrap, שמשמש כשרת האתחול ללקוחות Kafka שמתחברים מתוך אשכול Kubernetes. כל מאזיני Kafka הפנימיים זמינים בשירות הזה.
  • שירות ללא ממשק משתמש בשם my-cluster-kafka-brokers שמאפשר פענוח DNS של כתובות ה-IP של Pod של מתווך Kafka באופן ישיר. השירות הזה משמש לתקשורת בין ברוקרים.
  • שירות בשם my-cluster-zookeeper-client שמאפשר למתווכים של Kafka להתחבר לצומתי ZooKeeper כלקוחות.
  • שירות ללא ממשק משתמש בשם my-cluster-zookeeper-nodes, שמאפשר פענוח DNS של כתובות ה-IP של פוד ZooKeeper באופן ישיר. השירות הזה משמש לחיבור בין רפליקות של ZooKeeper.
  • פריסת Deployment בשם my-cluster-entity-operator שמכילה את topic-operator ו-user-operator ומאפשרת ניהול של משאבים מותאמים אישית KafkaTopics ו-KafkaUsers.

אפשר גם להגדיר שני NetworkPolicies כדי לאפשר קישוריות למאזיני Kafka מכל Pod ומרחב שמות. כללי המדיניות האלה גם יגבילו את החיבורים ל-ZooKeeper לברוקרים, ויאפשרו תקשורת בין ה-Pods של האשכול לבין יציאות שירות פנימיות שמוגבלות לתקשורת באשכול.

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

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

ב-Strimzi יש שיטה מותאמת ל-Kubernetes לניהול משתמשים, באמצעות User Operator נפרד והמשאב המותאם אישית התואם של Kubernetes,‏ KafkaUser, שמגדיר את הגדרות המשתמש. הגדרת המשתמש כוללת הגדרות לאימות ולהרשאה, ומקצה את המשתמש המתאים ב-Kafka.

‫Strimzi יכול ליצור מאזינים ומשתמשים של Kafka שתומכים בכמה מנגנוני אימות, כמו אימות שמבוסס על שם משתמש וסיסמה (SCRAM-SHA-512) או TLS. אפשר גם להשתמש באימות באמצעות OAuth 2.0, שנחשב בדרך כלל לגישה טובה יותר בהשוואה לשימוש בסיסמאות או באישורים לאימות, בגלל האבטחה וניהול פרטי הכניסה החיצוניים.

פריסת אשכול Kafka

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

  • אשכול Kafka עם אימות מבוסס-סיסמה (SCRAM-SHA-512) שמופעל באחד מהמאזינים.
  • KafkaTopic עם 3 העתקים.
  • KafkaUser עם רשימת ACL שבה מצוין שלמשתמש יש הרשאות קריאה וכתיבה בנושא.

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

    kubectl apply -n kafka -f kafka-strimzi/manifests/03-auth/my-cluster.yaml

  2. יוצרים Topic, User ו-Pod של לקוח כדי להריץ פקודות מול אשכול Kafka:

    kubectl apply -n kafka -f kafka-strimzi/manifests/03-auth/topic.yaml
kubectl apply -n kafka -f kafka-strimzi/manifests/03-auth/my-user.yaml

    ‫Secret my-user עם פרטי הכניסה של המשתמש מותקן ב-Pod של הלקוח כVolume.

    פרטי הכניסה האלה מאשרים שלמשתמש יש הרשאות לפרסם הודעות בנושא באמצעות המאזין, כשהאימות מבוסס הסיסמה (SCRAM-SHA-512) מופעל.

  3. יוצרים פוד לקוח:

    kubectl apply -n kafka -f kafka-strimzi/manifests/03-auth/kafkacat.yaml

  4. ממתינים כמה דקות עד שה-Pod של הלקוח יהפוך ל-Ready ואז מתחברים אליו:

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

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

    echo "Message from my-user" |kcat \
  -b my-cluster-kafka-bootstrap.kafka.svc.cluster.local:9094 \
  -X security.protocol=SASL_SSL \
  -X sasl.mechanisms=SCRAM-SHA-512 \
  -X sasl.username=my-user \
  -X sasl.password=$(cat /my-user/password) \
  -t my-topic -P
kcat -b my-cluster-kafka-bootstrap.kafka.svc.cluster.local:9094 \
  -X security.protocol=SASL_SSL \
  -X sasl.mechanisms=SCRAM-SHA-512 \
  -X sasl.username=my-user \
  -X sasl.password=$(cat /my-user/password) \
  -t my-topic -C

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

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

    מקלידים CTRL+C כדי לעצור את תהליך הצרכן.

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

    exit

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

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

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

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

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

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

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

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

המניפסט הבא מתאר StorageClass שמשתמש באפשרות reclaimPolicy של Retain:

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

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

# ...
spec:
  kafka:
    # ...
    storage:
      type: persistent-claim
      size: 100Gi
      class: premium-rwo-retain

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

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

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

הסרת המשאבים

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

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

    כדי למחוק פרויקט ב- 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)")

    השלב הזה נדרש כי כברירת מחדל, Strimzi משתמש בפרמטר deleteClaim: false לאחסון. אם מוחקים את האשכול, כל הדיסקים נשארים זמינים.

  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

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