פיתוח אפליקציית הפקה ב-Python

פיתוח אפליקציית הפקה ב-Python

במאמר הזה מוסבר איך לפתח אפליקציית Python producer שמבצעת אימות באמצעות Application Default Credentials (‏ADC) עם אשכול של שירות מנוהל ל-Apache Kafka. באמצעות ADC, אפליקציות שפועלות ב- Google Cloud יכולות למצוא באופן אוטומטי את פרטי הכניסה הנכונים ולאמת את עצמן בשירותי Google Cloud .

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

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

איך יוצרים אשכול

המסוף

1. עוברים אל הדף שירות מנוהל ל-Apache Kafka > Clusters.

   מעבר אל Clusters

2. לוחצים על יצירה.
3. בתיבה שם האשכול, מזינים שם לאשכול.
4. ברשימה Region, בוחרים מיקום לאשכול.
5. בקטע Network configuration (תצורת רשת), מגדירים את רשת המשנה שבה אפשר לגשת לאשכול:
   1. בקטע Project (פרויקט), בוחרים את הפרויקט.
   2. בקטע רשת, בוחרים את רשת ה-VPC.
   3. בקטע רשת משנה, בוחרים את רשת המשנה.
   4. לוחצים על סיום.
6. לוחצים על יצירה.

אחרי שלוחצים על יצירה, מצב האשכול הוא Creating. כשהאשכול מוכן, הסטטוס שלו הוא Active.

gcloud

כדי ליצור אשכול Kafka, מריצים את הפקודה managed-kafka clusters create.

gcloud managed-kafka clusters create KAFKA_CLUSTER \
--location=REGION \
--cpu=3 \
--memory=3GiB \
--subnets=projects/PROJECT_ID/regions/REGION/subnetworks/SUBNET_NAME \
--async

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

• ‫KAFKA_CLUSTER: שם לאשכול Kafka
• ‫REGION: המיקום של האשכול
• ‫PROJECT_ID: מזהה הפרויקט
• ‫SUBNET_NAME: תת-הרשת שבה רוצים ליצור את האשכול, לדוגמה default

מידע על מיקומים נתמכים זמין במאמר בנושא מיקומים של שירות מנוהל ל-Apache Kafka.

הפקודה פועלת באופן אסינכרוני ומחזירה מזהה פעולה:

Check operation [projects/PROJECT_ID/locations/REGION/operations/OPERATION_ID] for status.

כדי לעקוב אחרי ההתקדמות של פעולת היצירה, משתמשים בפקודה gcloud managed-kafka operations describe:

gcloud managed-kafka operations describe OPERATION_ID \
  --location=REGION

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

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

כדי לקבל את ההרשאות שדרושות ליצירה ולהגדרה של מכונת VM של לקוח, צריך לבקש מהאדמין להקצות לכם את תפקידי ה-IAM הבאים בפרויקט:

• Compute Instance Admin (v1) (roles/compute.instanceAdmin.v1)
• אדמין IAM בפרויקט (roles/resourcemanager.projectIamAdmin)
• צפייה בתפקיד (roles/iam.roleViewer)
• משתמש בחשבון שירות (roles/iam.serviceAccountUser)

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

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

הגדרת מכונה וירטואלית של לקוח

יוצרים מכונה וירטואלית (VM) של Linux ב-Compute Engine שיכולה לגשת לאשכול Kafka. כשמגדירים את המכונה הווירטואלית, צריך להגדיר את האפשרויות הבאות:

• אזור. יוצרים את המכונה הווירטואלית באותו אזור שבו נמצא אשכול Kafka.

• Subnet. יוצרים את המכונה הווירטואלית באותה רשת VPC כמו רשת המשנה שבה השתמשתם בהגדרת אשכול Kafka. מידע נוסף זמין במאמר בנושא הצגת רשתות המשנה של אשכול.

• היקפי הרשאות. מקצים את https://www.googleapis.com/auth/cloud-platform היקף הגישה למכונה הווירטואלית. היקף ההרשאות הזה מאשר למכונה הווירטואלית לשלוח בקשות אל Managed Kafka API.

בשלבים הבאים מוסבר איך מגדירים את האפשרויות האלה.

המסוף

1. נכנסים לדף Create an instance במסוף Google Cloud .

   יצירת מופע

2. בחלונית Machine configuration:

   1. בשדה Name, מציינים שם למכונה. מידע נוסף זמין במאמר מוסכמות למתן שמות למשאבים.

   2. ברשימה Region, בוחרים את אותו אזור כמו אשכול Kafka.

   3. ברשימה Zone, בוחרים אזור.

3. בתפריט הניווט, לוחצים על Networking (רשת). בחלונית Networking שמופיעה, מבצעים את הפעולות הבאות:

   1. עוברים לקטע Network interfaces.

   2. כדי להרחיב את ממשק ברירת המחדל של הרשת, לוחצים על החץ .

   3. בשדה Network, בוחרים את רשת ה-VPC.

   4. ברשימה Subnetwork, בוחרים את רשת המשנה.

   5. לוחצים על סיום.

4. בתפריט הניווט, לוחצים על אבטחה. בחלונית Security שמופיעה, מבצעים את הפעולות הבאות:

   1. בקטע Access scopes בוחרים באפשרות Set access for each API.

   2. ברשימת היקפי הגישה, מחפשים את הרשימה הנפתחת Cloud Platform ובוחרים באפשרות Enabled.

5. לוחצים על יצירה כדי ליצור את המכונה הווירטואלית.

gcloud

כדי ליצור את מכונת ה-VM, משתמשים בפקודה gcloud compute instances create.

gcloud compute instances create VM_NAME \
  --scopes=https://www.googleapis.com/auth/cloud-platform \
  --subnet=projects/PROJECT_ID/regions/REGION/subnetworks/SUBNET \
  --zone=ZONE

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

• ‫VM_NAME: השם של ה-VM
• ‫PROJECT_ID: מזהה הפרויקט
• ‫REGION: האזור שבו יצרתם את אשכול Kafka, לדוגמה us-central1
• ‫SUBNET: תת-רשת באותה רשת VPC כמו תת-הרשת שבה השתמשתם בתצורת האשכול
• ‫ZONE: אזור באזור שבו יצרתם את האשכול, למשל us-central1-c

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

מתן תפקידים ב-IAM

מקצים את התפקידים הבאים בניהול זהויות והרשאות גישה (IAM) לחשבון השירות שמוגדר כברירת מחדל ב-Compute Engine:

• ‫Managed Kafka Client‏ (roles/managedkafka.client)
• יצירת אסימונים בחשבון שירות (roles/iam.serviceAccountTokenCreator)

• יצירת אסימונים מסוג OpenID בחשבון שירות (roles/iam.serviceAccountOpenIdTokenCreator)

המסוף

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

   כניסה לדף IAM

2. מחפשים את השורה של חשבון השירות של Compute Engine שמוגדר כברירת מחדל ולוחצים על עריכת החשבון הראשי.

3. לוחצים על Add another role ובוחרים את התפקיד Managed Kafka Client. חוזרים על השלב הזה עבור התפקידים יצירת אסימונים בחשבון שירות ויצירת אסימונים מסוג OpenID בחשבון שירות.

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

gcloud

כדי להקצות תפקידים ב-IAM, משתמשים בפקודה gcloud projects add-iam-policy-binding.

gcloud projects add-iam-policy-binding PROJECT_ID \
  --member="serviceAccount:[e-mail]" \
  --role=roles/managedkafka.client

gcloud projects add-iam-policy-binding PROJECT_ID\
  --member="serviceAccount:[e-mail]" \
  --role=roles/iam.serviceAccountTokenCreator

gcloud projects add-iam-policy-binding PROJECT_ID \
  --member="serviceAccount:[e-mail]" \
  --role=roles/iam.serviceAccountOpenIdTokenCreator

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

• ‫PROJECT_ID: מזהה הפרויקט

• ‫PROJECT_NUMBER: מספר הפרויקט

כדי לקבל את מספר הפרויקט, מריצים את הפקודה gcloud projects describe:

gcloud projects describe PROJECT_ID

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

התחברות למכונה הווירטואלית

משתמשים ב-SSH כדי להתחבר למופע של המכונה הווירטואלית.

המסוף

1. נכנסים לדף VM instances.

   כניסה לדף VM instances

2. ברשימת המכונות הווירטואליות, מוצאים את שם המכונה הווירטואלית ולוחצים על SSH.

gcloud

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

gcloud compute ssh VM_NAME \
  --project=PROJECT_ID \
  --zone=ZONE

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

• ‫VM_NAME: השם של ה-VM
• ‫PROJECT_ID: מזהה הפרויקט
• ‫ZONE: האזור שבו יצרתם את המכונה הווירטואלית

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

יצירת אפליקציית הפקה ב-Python

בסשן ה-SSH, מריצים את הפקודות הבאות כדי ליצור אפליקציית הפקה.

1. מתקינים את pip, מנהל חבילות Python ומנהל הסביבה הווירטואלית:

   sudo apt install python3-pip -y
   sudo apt install python3-venv -y
   

2. יוצרים סביבה וירטואלית חדשה (venv) ומפעילים אותה:

   python3 -m venv kafka
   source kafka/bin/activate
   

3. מתקינים את לקוח confluent-kafka ואת יחסי התלות האחרים:

   pip install confluent-kafka google-auth urllib3 packaging requests
   

4. מעתיקים את קוד הלקוח הבא של היצרן לקובץ בשם producer.py

   import confluent_kafka
   import argparse
   from tokenprovider import TokenProvider
   
   parser = argparse.ArgumentParser()
   parser.add_argument('-b', '--bootstrap-servers', dest='bootstrap', type=str, required=True)
   parser.add_argument('-t', '--topic-name', dest='topic_name', type=str, default='example-topic', required=False)
   parser.add_argument('-n', '--num_messages', dest='num_messages', type=int, default=1, required=False)
   args = parser.parse_args()
   
   token_provider = TokenProvider()
   
   config = {
       'bootstrap.servers': args.bootstrap,
       'security.protocol': 'SASL_SSL',
       'sasl.mechanisms': 'OAUTHBEARER',
       'oauth_cb': token_provider.get_token,
   }
   
   producer = confluent_kafka.Producer(config)
   
   def callback(error, message):
       if error is not None:
           print(error)
           return
       print("Delivered a message to {}[{}]".format(message.topic(), message.partition()))
   
   for i in range(args.num_messages):
   
     message = f"{i} hello world!".encode('utf-8')
     producer.produce(args.topic_name, message, callback=callback)
   
   producer.flush()
   

5. עכשיו צריך להטמיע את ספק אסימוני ה-OAuth. שומרים את הקוד הבא בקובץ בשם tokenprovider.py:

   import base64
   import datetime
   import http.server
   import json
   import google.auth
   from google.auth.transport.urllib3 import Request
   import urllib3
   import time
   
   def encode(source):
     """Safe base64 encoding."""
     return base64.urlsafe_b64encode(source.encode('utf-8')).decode('utf-8').rstrip('=')
   
   class TokenProvider(object):
     """
     Provides OAuth tokens from Google Cloud Application Default credentials.
     """
     HEADER = json.dumps({'typ':'JWT', 'alg':'GOOG_OAUTH2_TOKEN'})
   
     def __init__(self, **config):
       self.credentials, _project = google.auth.default()
       self.http_client = urllib3.PoolManager()
   
     def get_credentials(self):
       if not self.credentials.valid:
         self.credentials.refresh(Request(self.http_client))
       return self.credentials
   
     def get_jwt(self, creds):
       token_data = dict(
         exp=creds.expiry.replace(tzinfo=datetime.timezone.utc).timestamp(),
         iat=datetime.datetime.now(datetime.timezone.utc).timestamp(),
         iss='Google',
         sub=creds.service_account_email
       )
       return json.dumps(token_data)
   
     def get_token(self, args):
       creds = self.get_credentials()
       token = '.'.join([
         encode(self.HEADER),
         encode(self.get_jwt(creds)),
         encode(creds.token)
       ])
   
       # compute expiry time
       expiry_utc = creds.expiry.replace(tzinfo=datetime.timezone.utc)
       now_utc = datetime.datetime.now(datetime.timezone.utc)
       expiry_seconds = (expiry_utc - now_utc).total_seconds()
   
       return token, time.time() + expiry_seconds
   

6. עכשיו אפשר להפעיל את האפליקציה:

   python producer.py -b bootstrap.CLUSTER_ID.REGION.managedkafka.PROJECT_ID.cloud.goog:9092
   

הסרת המשאבים

כדי לא לצבור חיובים לחשבון Google Cloud על המשאבים שבהם השתמשתם בדף הזה, פועלים לפי השלבים הבאים:

המסוף

1. מוחקים את מכונת ה-VM.

   1. נכנסים לדף VM instances.

      כניסה לדף VM instances

   2. בוחרים את המכונה הווירטואלית ולוחצים על מחיקה.

2. מוחקים את אשכול Kafka.

   1. עוברים לדף שירות מנוהל ל-Apache Kafka > Clusters (אשכולות).

      מעבר אל Clusters

   2. בוחרים את אשכול Kafka ולוחצים על מחיקה.

gcloud

1. כדי למחוק את מכונת ה-VM, משתמשים בפקודה gcloud compute instances delete.

   gcloud compute instances delete VM_NAME --zone=ZONE
   

2. כדי למחוק את אשכול Kafka, משתמשים בפקודה gcloud managed-kafka clusters delete.

   gcloud managed-kafka clusters delete CLUSTER_ID \
     --location=REGION --async
   

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

• סקירה כללית על השירות המנוהל ל-Apache Kafka

• אימות של לקוחות Kafka באמצעות שירות מנוהל ל-Apache Kafka

• כלי אימות ומסמכים בנושא שירות מנוהל ל-Apache Kafka

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