Google uses AI technology to translate content into your preferred language. AI translations can contain errors.

Développer une application productrice Python

Découvrez comment développer une application productrice Python qui s'authentifie auprès d'un cluster Managed Service pour Apache Kafka à l'aide des identifiants par défaut de l'application (ADC). Les ADC permettent aux applications exécutées sur Google Cloud de trouver et d' utiliser automatiquement les identifiants appropriés pour s'authentifier auprès des Google Cloud services.

Avant de commencer

Avant de commencer ce tutoriel, créez un cluster Managed Service pour Apache Kafka. Si vous disposez déjà d'un cluster, vous pouvez ignorer cette étape.

Créer un cluster

Console

Accédez à la page Managed Service pour Apache Kafka > Clusters.
Accéder aux clusters
Cliquez sur Créer.
Dans la zone Nom du cluster, saisissez un nom pour le cluster.
Dans la liste Région, sélectionnez un emplacement pour le cluster.
Pour Configuration réseau, configurez le sous-réseau où le cluster est accessible :
1. Pour Project (Projet), sélectionnez votre projet.
2. Sous Network (Réseau), sélectionnez le réseau VPC.
3. Pour Subnet (Sous-réseau), sélectionnez le sous-réseau.
4. Cliquez sur OK.
Cliquez sur Créer.

Une fois que vous avez cliqué sur Créer, l'état du cluster est Creating. Lorsque le cluster est prêt, l'état est Active.

gcloud

Pour créer un cluster Kafka, exécutez la managed-kafka clusters create commande.

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

Remplacez les éléments suivants :

KAFKA_CLUSTER : nom du cluster Kafka.
REGION : emplacement du cluster.
PROJECT_ID : ID du projet.
SUBNET_NAME : sous-réseau dans lequel vous souhaitez créer le cluster, par exemple default.

Pour en savoir plus sur les emplacements compatibles, consultez la section Emplacements Managed Service pour Apache Kafka.

La commande s'exécute de manière asynchrone et renvoie un ID d'opération :

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

Pour suivre la progression de l'opération de création, utilisez la gcloud managed-kafka operations describe commande :

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

Lorsque le cluster est prêt, la sortie de cette commande inclut l'entrée state: ACTIVE. Pour en savoir plus, consultez la section Surveiller l'opération de création du cluster.

Rôles requis

Pour obtenir les autorisations nécessaires pour créer et configurer une VM cliente, demandez à votre administrateur de vous accorder les rôles IAM suivants sur le projet :

Administrateur d'instances Compute (v1) (roles/compute.instanceAdmin.v1)
Administrateur de projet IAM (roles/resourcemanager.projectIamAdmin)
Lecteur de rôles (roles/iam.roleViewer)
Utilisateur du compte de service (roles/iam.serviceAccountUser)

Pour en savoir plus sur l'attribution de rôles, consultez la page Gérer l'accès aux projets, aux dossiers et aux organisations.

Vous pouvez également obtenir les autorisations requises avec des rôles personnalisés ou d'autres rôles prédéfinis.

Configurer une VM cliente

Créez une instance de machine virtuelle (VM) Linux dans Compute Engine qui peut accéder au cluster Kafka. Lorsque vous configurez la VM, définissez les options suivantes :

Région : créez la VM dans la même région que votre cluster Kafka.
Sous-réseau : créez la VM dans le même réseau VPC que le sous-réseau que vous avez utilisé dans la configuration de votre cluster Kafka. Pour en savoir plus, consultez la section Afficher les sous-réseaux d'un cluster.
Niveaux d'accès : attribuez le https://www.googleapis.com/auth/cloud-platform niveau d'accès à la VM. Ce niveau autorise la VM à envoyer des requêtes à l'API Managed Kafka.

Les étapes suivantes montrent comment définir ces options.

Console

Accédez à la page Créer une instance dans la Google Cloud console.

Créer une instance
Dans le volet Configuration de la machine, procédez comme suit :
1. Dans le champ Nom, spécifiez un nom pour votre instance. Pour en savoir plus, consultez Convention d'attribution de noms aux ressources.
2. Dans la liste Région, sélectionnez la même région que votre cluster Kafka.
3. Dans la liste Zone, sélectionnez une zone.
Dans le menu de navigation, cliquez sur Mise en réseau. Dans le volet Mise en réseau qui s'affiche, procédez comme suit :
1. Accédez à la section Interfaces réseau.
2. Pour développer l'interface réseau par défaut, cliquez sur la flèche.
3. Dans le champ Réseau, sélectionnez le réseau VPC.
4. Dans la liste Sous-réseau, sélectionnez le sous-réseau.
5. Cliquez sur OK.
Dans le menu de navigation, cliquez sur Sécurité. Dans le volet Sécurité qui s'affiche, procédez comme suit :
1. Pour Niveaux d'accès, sélectionnez Définir l'accès pour chaque API.
2. Dans la liste des niveaux d'accès, recherchez la liste déroulante Cloud Platform et sélectionnez Activé.
Cliquez sur Créer pour créer la VM.

gcloud

Pour créer l'instance de VM, utilisez la gcloud compute instances create commande.

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

Remplacez les éléments suivants :

VM_NAME : nom de la VM.
PROJECT_ID : ID du projet.
REGION : région dans laquelle vous avez créé le cluster Kafka, par exemple us-central1
SUBNET : sous-réseau dans le même réseau VPC que le sous-réseau que vous avez utilisé dans la configuration du cluster.
ZONE : zone dans la région où vous avez créé le cluster, par exemple us-central1-c.

Pour en savoir plus sur la création d'une VM, consultez la section Créer une instance de VM dans un sous-réseau spécifique.

Accorder des rôles IAM

Attribuez les rôles IAM (Identity and Access Management) suivants au compte de service Compute Engine par défaut :

Client Managed Kafka (roles/managedkafka.client)
Créateur de jetons du compte de service (roles/iam.serviceAccountTokenCreator)
Créateur de jetons OpenID du compte de service (roles/iam.serviceAccountOpenIdTokenCreator)

Console

Dans la console Google Cloud , accédez à la page IAM.

Accéder à IAM
Recherchez la ligne Compte de service Compute Engine par défaut , puis cliquez sur Modifier le compte principal.
Cliquez sur Ajouter un autre rôle , puis sélectionnez le rôle Client Managed Kafka. Répétez cette étape pour les rôles Créateur de jetons du compte de service et Créateur de jetons OpenID du compte de service.
Cliquez sur Enregistrer.

gcloud

Pour accorder des rôles IAM, utilisez la gcloud projects add-iam-policy-binding commande.

gcloud projects add-iam-policy-binding PROJECT_ID \
  --member="serviceAccount:PROJECT_NUMBER-compute@developer.gserviceaccount.com" \
  --role=roles/managedkafka.client

gcloud projects add-iam-policy-binding PROJECT_ID\
  --member="serviceAccount:PROJECT_NUMBER-compute@developer.gserviceaccount.com" \
  --role=roles/iam.serviceAccountTokenCreator

gcloud projects add-iam-policy-binding PROJECT_ID \
  --member="serviceAccount:PROJECT_NUMBER-compute@developer.gserviceaccount.com" \
  --role=roles/iam.serviceAccountOpenIdTokenCreator

Remplacez les éléments suivants :

PROJECT_ID : ID du projet.
PROJECT_NUMBER : numéro de votre projet.

Pour obtenir le numéro de projet, exécutez la gcloud projects describe commande :

gcloud projects describe PROJECT_ID

Pour en savoir plus, consultez la section Trouver le nom, le numéro et l'ID du projet.

Se connecter à la VM

Utilisez SSH pour vous connecter à l'instance de VM.

Console

Accédez à la page Instances de VM.

Accéder à la page Instances de VM
Dans la liste des instances de VM, recherchez le nom de la VM, puis cliquez sur SSH.

gcloud

Pour vous connecter à la VM, utilisez la gcloud compute ssh commande.

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

Remplacez les éléments suivants :

VM_NAME : nom de la VM.
PROJECT_ID : ID du projet.
ZONE : zone dans laquelle vous avez créé la VM.

Une configuration supplémentaire peut être requise lors de la première utilisation de SSH. Pour en savoir plus, consultez la section À propos des connexions SSH.

Créer une application productrice Python

À partir de votre session SSH, exécutez les commandes suivantes pour créer une application productrice.

Installez pip, un gestionnaire de packages Python et le gestionnaire d'environnement virtuel :
```
sudo apt install python3-pip -y
sudo apt install python3-venv -y
```

Créez un environnement virtuel (venv) et activez-le :

python3 -m venv kafka
source kafka/bin/activate

Installez le client confluent-kafka et d'autres dépendances :

pip install confluent-kafka google-auth urllib3 packaging requests

Copiez le code client producteur suivant dans un fichier nommé 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()

Vous avez maintenant besoin d'une implémentation du fournisseur de jetons OAuth. Enregistrez le code suivant dans un fichier nommé 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

Recherchez l'adresse d'amorçage de votre cluster Kafka en suivant Afficher l'adresse d'amorçage d'un cluster
Vous êtes maintenant prêt à exécuter l'application :
```
python producer.py -b BOOTSTRAP
```

Libérer de l'espace

Pour éviter que les ressources utilisées dans cette démonstration soient facturées sur votre Google Cloud compte pour les ressources utilisées sur cette page, procédez comme suit :

Console

Supprimez l'instance de VM.
1. Accédez à la page Instances de VM.
  
  Accéder à la page Instances de VM
2. Sélectionnez la VM, puis cliquez sur Supprimer.
Supprimez le cluster Kafka.
1. Accédez à la page Managed Service pour Apache Kafka > Clusters.
  
  Accéder aux clusters
2. Sélectionnez le cluster Kafka, puis cliquez sur Supprimer.

gcloud

Pour supprimer la VM, utilisez la gcloud compute instances delete commande.
```
gcloud compute instances delete VM_NAME --zone=ZONE
```
Pour supprimer le cluster Kafka, utilisez la gcloud managed-kafka clusters delete commande.
```
gcloud managed-kafka clusters delete CLUSTER_ID \
  --location=REGION --async
```

Étape suivante

Apache Kafka® est une marque déposée d'Apache Software Foundation ou de ses filiales aux États-Unis et/ou dans d'autres pays.