Créer un sujet d'importation Amazon Kinesis Data Streams

Un sujet d'importation Amazon Kinesis Data Streams vous permet d'ingérer en continu des données à partir d'Amazon Kinesis Data Streams en tant que source externe et dans Pub/Sub. Vous pouvez ensuite diffuser les données vers n'importe quelle destination compatible avec Pub/Sub.

Ce document explique comment créer et gérer des sujets d'importation Amazon Kinesis Data Streams. Pour créer un thème standard, consultez la section Créer un thème standard.

Avant de commencer

• En savoir plus sur le processus de publication Pub/Sub

• Configurez les rôles et autorisations requis pour gérer les sujets d'importation Amazon Kinesis Data Streams, y compris les suivants:

  • Ajouter le rôle Éditeur Pub/Sub au compte de service Pub/Sub

  • Ajouter le rôle "Utilisateur du compte de service" au compte de service

• Configurez la fédération d'identité de charge de travail pour que Google Cloud puisse accéder au service de streaming externe.

Rôles et autorisations requis pour gérer les sujets d'importation

Pour obtenir les autorisations nécessaires pour créer et gérer des sujets d'importation Amazon Kinesis Data Streams, demandez à votre administrateur de vous accorder le rôle IAM Éditeur Pub/Sub(roles/pubsub.editor) sur votre sujet ou votre projet. 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.

Ce rôle prédéfini contient les autorisations requises pour créer et gérer des sujets d'importation Amazon Kinesis Data Streams. Pour connaître les autorisations exactes requises, développez la section Autorisations requises :

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

Vous pouvez configurer le contrôle des accès au niveau du projet et au niveau de ressources individuelles.

Configurer une identité fédérée pour accéder à Kinesis Data Streams

La fédération d'identité de charge de travail permet aux services Google Cloud d'accéder aux charges de travail exécutées en dehors de Google Cloud. Avec la fédération d'identité, vous n'avez pas besoin de gérer ni de transmettre des identifiants à Google Cloud pour accéder à vos ressources dans d'autres clouds. À la place, vous pouvez utiliser les identités des charges de travail elles-mêmes pour vous authentifier auprès de Google Cloud et accéder aux ressources.

Créer un compte de service dans Google Cloud

Il s'agit d'une étape facultative. Si vous disposez déjà d'un compte de service, vous pouvez l'utiliser dans cette procédure au lieu d'en créer un autre. Si vous utilisez un compte de service existant, consultez la section Enregistrer l'ID unique du compte de service pour l'étape suivante.

Pour un sujet d'importation Amazon Kinesis Data Streams, Pub/Sub utilise le compte de service comme identité pour accéder aux ressources d'AWS.

Pour en savoir plus sur la création d'un compte de service, y compris les conditions préalables, les rôles et autorisations requis, ainsi que les consignes de dénomination, consultez la page Créer des comptes de service. Une fois le compte de service créé, vous devrez peut-être attendre au moins 60 secondes avant de l'utiliser. Ce comportement se produit car les opérations de lecture sont cohérentes à terme. Il peut s'écouler un certain temps avant que le nouveau compte de service devienne visible.

Enregistrer l'ID unique du compte de service

Vous avez besoin d'un ID unique de compte de service pour configurer un rôle dans la console AWS.

1. Dans la console Google Cloud, accédez à la page d'informations du compte de service.

   Accéder au compte de service

2. Cliquez sur le compte de service que vous venez de créer ou celui que vous prévoyez d'utiliser.

3. Sur la page Détails du compte de service, notez l'ID unique.

   Vous aurez besoin de cet ID dans la section Créer un rôle dans AWS à l'aide d'une stratégie de confiance personnalisée.

Ajouter le rôle Créateur de jetons du compte de service au compte de service Pub/Sub

Le rôle Créateur de jetons du compte de service (roles/iam.serviceAccountTokenCreator) permet aux comptes principaux de créer des identifiants éphémères pour un compte de service. Ces jetons ou identifiants sont utilisés pour usurper l'identité du compte de service.

Pour plus d'informations sur l'emprunt de l'identité d'un compte de service, consultez la page Emprunter l'identité d'un compte de service.

Vous pouvez également ajouter le rôle Éditeur Pub/Sub (roles/pubsub.publisher) au cours de cette procédure. Pour en savoir plus sur ce rôle et pourquoi vous l'ajoutez, consultez Ajouter le rôle d'éditeur Pub/Sub au compte de service Pub/Sub.

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

   Accéder à IAM

2. Activez l'option Inclure les attributions de rôles fournies par Google.

3. Recherchez le compte de service au format service-{PROJECT_NUMBER}@gcp-sa-pubsub.iam.gserviceaccount.com.

4. Pour ce compte de service, cliquez sur le bouton Modifier le compte principal.

5. Si nécessaire, cliquez sur Ajouter un autre rôle.

6. Recherchez et sélectionnez le rôle Créateur de jetons du compte de service (roles/iam.serviceAccountTokenCreator).

7. Cliquez sur Enregistrer.

Créer une stratégie dans AWS

Vous avez besoin d'une stratégie dans AWS pour autoriser Pub/Sub à s'authentifier auprès d'AWS afin qu'il puisse ingérer des données à partir d'un flux de données AWS Kinesis. Avant de créer une stratégie AWS, créez un flux de données Kinesis et un consommateur enregistré dessus. Nous vous recommandons de procéder ainsi afin de pouvoir limiter les autorisations au flux spécifique.

• Pour en savoir plus sur la création d'un flux de données AWS Kinesis, consultez la section Flux de données Kinesis.

• Pour en savoir plus sur l'API AWS Kinesis Data Streams utilisée pour enregistrer des consommateurs, consultez RegisterStreamConsumer.

• Pour découvrir d'autres méthodes et obtenir des informations sur la création d'une stratégie dans AWS, consultez la section Créer des stratégies IAM.

Pour créer une règle dans AWS, procédez comme suit:

1. Connectez-vous à la console de gestion AWS et ouvrez la console IAM.

2. Dans le volet de navigation de la console IAM, cliquez sur Gestion des accès > Règles.

3. Cliquez sur Créer une règle.

4. Pour Sélectionner un service, sélectionnez Kinesis.

5. Pour action autorisée, sélectionnez les éléments suivants:

   • List > ListShards.

     Cette action permet d'autoriser la liste des fragments d'un flux et fournit des informations sur chaque fragment.

   • Lire > SubscribeToShard.

     Cette action autorise l'écoute d'un segment spécifique avec un fan-out amélioré.

   • Lire > DescribeStreamConsumer

     Cette action permet d'obtenir la description d'un consommateur de flux enregistré.

   Ces autorisations couvrent la lecture à partir du flux. Pub/Sub n'est compatible qu'avec la lecture à partir d'un flux Kinesis avec diffusion étendue à l'aide de l'API SubscribeToShard de streaming.

6. Pour Ressources, si vous souhaitez limiter la règle à un flux ou à un client spécifiques (recommandé), spécifiez l'ARN du client et l'ARN du flux.

7. Cliquez sur Ajouter des autorisations.

8. Pour Sélectionnez un service, saisissez STS, puis sélectionnez-le.

9. Pour action allowed (Action autorisée), sélectionnez Write > AssumeRoleWithWebIdentity (Écrire > AssumeRoleWithWebIdentity).

   Cette action autorise l'obtention d'un ensemble d'identifiants de sécurité temporaires pour que Pub/Sub s'authentifie auprès du flux de données Kinesis à l'aide de la fédération d'identités.

10. Cliquez sur Suivant.

11. Saisissez un nom et une description pour la règle.

12. Cliquez sur Créer une règle.

Créer un rôle dans AWS à l'aide d'une stratégie de confiance personnalisée

Vous devez créer un rôle dans AWS afin que Pub/Sub puisse s'authentifier auprès d'AWS pour ingérer les données de Kinesis Data Streams.

Pour créer un rôle à l'aide d'une stratégie de confiance personnalisée, procédez comme suit:

1. Connectez-vous à la console de gestion AWS et ouvrez la console IAM.

2. Dans le volet de navigation de la console IAM, cliquez sur Roles (Rôles).

3. Cliquez sur Créer un rôle.

4. Dans le champ Sélectionner l'entité de confiance, sélectionnez Règle d'approbation personnalisée.

5. Dans la section Règle de confiance personnalisée, saisissez ou collez ce qui suit:

   {
     "Version": "2012-10-17",
     "Statement": [
       {
        "Effect": "Allow",
        "Principal": {
           "Federated": "accounts.google.com"
        },
        "Action": "sts:AssumeRoleWithWebIdentity",
        "Condition": {
            "StringEquals": {
              "accounts.google.com:sub": "<SERVICE_ACCOUNT_UNIQUE_ID>"
            }
         }
       }
     ]
   }
   

   Remplacez <SERVICE_ACCOUNT_UNIQUE_ID> par l'ID unique du compte de service que vous avez enregistré dans Enregistrer l'ID unique du compte de service.

6. Cliquez sur Suivant.

7. Pour Ajouter des autorisations, recherchez et sélectionnez la règle personnalisée que vous venez de créer.

8. Cliquez sur Suivant.

9. Saisissez un nom et une description du rôle.

10. Cliquez sur Créer un rôle.

Ajouter le rôle Éditeur Pub/Sub au compte de service Pub/Sub

Vous devez attribuer un rôle d'éditeur au compte de service Pub/Sub afin que Pub/Sub puisse publier sur le sujet d'importation à partir d'AWS Kinesis Data Streams.

• Pour activer la publication à partir de tous les sujets d'un projet, consultez Activer la publication à partir de tous les sujets. Utilisez cette méthode si vous n'avez pas créé de sujets d'importation AWS Kinesis Data Streams.

• Pour activer la publication à partir d'un sujet spécifique (recommandé), consultez Activer la publication à partir d'un seul sujet. N'utilisez cette méthode que si le sujet d'importation AWS Kinesis Data Streams existe déjà.

Activer la publication à partir de tous les thèmes

Utilisez cette méthode si vous n'avez pas créé de sujets d'importation AWS Kinesis Data Streams.

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

   Accéder à IAM

2. Activez l'option Inclure les attributions de rôles fournies par Google.

3. Recherchez le compte de service au format service-{PROJECT_NUMBER}@gcp-sa-pubsub.iam.gserviceaccount.com.

4. Pour ce compte de service, cliquez sur le bouton Modifier le compte principal.

5. Si nécessaire, cliquez sur Ajouter un autre rôle.

6. Recherchez et sélectionnez le rôle Éditeur Pub/Sub (roles/pubsub.publisher).

7. Cliquez sur Enregistrer.

Activer la publication à partir d'un seul sujet

N'utilisez cette méthode que si le sujet d'importation AWS Kinesis Data Streams existe déjà.

1. In the Google Cloud console, activate Cloud Shell.

   Activate Cloud Shell

   At the bottom of the Google Cloud console, a Cloud Shell session starts and displays a command-line prompt. Cloud Shell is a shell environment with the Google Cloud CLI already installed and with values already set for your current project. It can take a few seconds for the session to initialize.

2. Exécutez la commande gcloud pubsub topics add-iam-policy-binding :

   gcloud pubsub topics add-iam-policy-binding TOPIC_ID\
      --member="serviceAccount:service-PROJECT_NUMBER@gcp-sa-pubsub.iam.gserviceaccount.com"\
      --role="roles/pubsub.publisher"

   Remplacez les éléments suivants :

   • TOPIC_ID correspond à l'ID du sujet d'importation Amazon Kinesis Data Streams.

   • PROJECT_NUMBER est le numéro du projet. Pour afficher le numéro de projet, consultez Identifier des projets.

Ajouter le rôle Utilisateur du compte de service au compte de service

Le rôle Utilisateur du compte de service (roles/iam.serviceAccountUser) inclut l'autorisation iam.serviceAccounts.actAs qui permet à un principal d'associer un compte de service aux paramètres d'ingestion du sujet d'importation Amazon Kinesis Data Streams et d'utiliser ce compte de service pour l'identité fédérée.

Procédez comme suit :

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

   Accéder à IAM

2. Pour le compte principal qui émet les appels de création ou de mise à jour de sujets, cliquez sur le bouton Modifier le compte principal.

3. Si nécessaire, cliquez sur Ajouter un autre rôle.

4. Recherchez et sélectionnez le rôle Utilisateur du compte de service (roles/iam.serviceAccountUser).

5. Cliquez sur Enregistrer.

Créer un sujet d'importation Amazon Kinesis Data Streams

Pour en savoir plus sur les propriétés associées à un sujet, consultez la section Propriétés d'un sujet.

Assurez-vous d'avoir effectué les procédures suivantes:

• Configurer une identité fédérée pour accéder à Kinesis Data Streams

• Ajouter le rôle Éditeur Pub/Sub au compte de service Pub/Sub

• Ajouter le rôle "Utilisateur du compte de service" au compte de service

Créer le sujet et l'abonnement séparément, même si vous le faites rapidement, peut entraîner une perte de données. Il existe une courte période pendant laquelle le sujet existe sans abonnement. Si des données sont envoyées au sujet pendant cette période, elles sont perdues. En créant d'abord le sujet, puis l'abonnement, puis en convertissant le sujet en sujet d'importation, vous vous assurez qu'aucun message n'est manqué lors du processus d'importation.

Pour créer un sujet d'importation Amazon Kinesis Data Streams, procédez comme suit:

import (
	"context"
	"fmt"
	"io"

	"cloud.google.com/go/pubsub"
)

func createTopicWithKinesisIngestion(w io.Writer, projectID, topicID string) error {
	// projectID := "my-project-id"
	// topicID := "my-topic"
	streamARN := "stream-arn"
	consumerARN := "consumer-arn"
	awsRoleARN := "aws-role-arn"
	gcpServiceAccount := "gcp-service-account"

	ctx := context.Background()
	client, err := pubsub.NewClient(ctx, projectID)
	if err != nil {
		return fmt.Errorf("pubsub.NewClient: %w", err)
	}
	defer client.Close()

	cfg := &pubsub.TopicConfig{
		IngestionDataSourceSettings: &pubsub.IngestionDataSourceSettings{
			Source: &pubsub.IngestionDataSourceAWSKinesis{
				StreamARN:         streamARN,
				ConsumerARN:       consumerARN,
				AWSRoleARN:        awsRoleARN,
				GCPServiceAccount: gcpServiceAccount,
			},
		},
	}
	t, err := client.CreateTopicWithConfig(ctx, topicID, cfg)
	if err != nil {
		return fmt.Errorf("CreateTopic: %w", err)
	}
	fmt.Fprintf(w, "Kinesis topic created: %v\n", t)
	return nil
}

import com.google.cloud.pubsub.v1.TopicAdminClient;
import com.google.pubsub.v1.IngestionDataSourceSettings;
import com.google.pubsub.v1.Topic;
import com.google.pubsub.v1.TopicName;
import java.io.IOException;

public class CreateTopicWithKinesisIngestionExample {
  public static void main(String... args) throws Exception {
    // TODO(developer): Replace these variables before running the sample.
    String projectId = "your-project-id";
    String topicId = "your-topic-id";
    // Kinesis ingestion settings.
    String streamArn = "stream-arn";
    String consumerArn = "consumer-arn";
    String awsRoleArn = "aws-role-arn";
    String gcpServiceAccount = "gcp-service-account";

    createTopicWithKinesisIngestionExample(
        projectId, topicId, streamArn, consumerArn, awsRoleArn, gcpServiceAccount);
  }

  public static void createTopicWithKinesisIngestionExample(
      String projectId,
      String topicId,
      String streamArn,
      String consumerArn,
      String awsRoleArn,
      String gcpServiceAccount)
      throws IOException {
    try (TopicAdminClient topicAdminClient = TopicAdminClient.create()) {
      TopicName topicName = TopicName.of(projectId, topicId);

      IngestionDataSourceSettings.AwsKinesis awsKinesis =
          IngestionDataSourceSettings.AwsKinesis.newBuilder()
              .setStreamArn(streamArn)
              .setConsumerArn(consumerArn)
              .setAwsRoleArn(awsRoleArn)
              .setGcpServiceAccount(gcpServiceAccount)
              .build();
      IngestionDataSourceSettings ingestionDataSourceSettings =
          IngestionDataSourceSettings.newBuilder().setAwsKinesis(awsKinesis).build();

      Topic topic =
          topicAdminClient.createTopic(
              Topic.newBuilder()
                  .setName(topicName.toString())
                  .setIngestionDataSourceSettings(ingestionDataSourceSettings)
                  .build());

      System.out.println("Created topic with Kinesis ingestion settings: " + topic.getAllFields());
    }
  }
}

/**
 * TODO(developer): Uncomment these variables before running the sample.
 */
// const topicNameOrId = 'YOUR_TOPIC_NAME_OR_ID';
// const roleArn = 'arn:aws:iam:...';
// const gcpServiceAccount = 'ingestion-account@...';
// const streamArn = 'arn:aws:kinesis:...';
// const consumerArn = 'arn:aws:kinesis:...';

// Imports the Google Cloud client library
const {PubSub} = require('@google-cloud/pubsub');

// Creates a client; cache this for further use
const pubSubClient = new PubSub();

async function createTopicWithKinesisIngestion(
  topicNameOrId,
  awsRoleArn,
  gcpServiceAccount,
  streamArn,
  consumerArn
) {
  // Creates a new topic with Kinesis ingestion.
  await pubSubClient.createTopic({
    name: topicNameOrId,
    ingestionDataSourceSettings: {
      awsKinesis: {
        awsRoleArn,
        gcpServiceAccount,
        streamArn,
        consumerArn,
      },
    },
  });
  console.log(`Topic ${topicNameOrId} created with AWS Kinesis ingestion.`);
}

from google.cloud import pubsub_v1
from google.pubsub_v1.types import Topic
from google.pubsub_v1.types import IngestionDataSourceSettings

# TODO(developer)
# project_id = "your-project-id"
# topic_id = "your-topic-id"
# stream_arn = "your-stream-arn"
# consumer_arn = "your-consumer-arn"
# aws_role_arn = "your-aws-role-arn"
# gcp_service_account = "your-gcp-service-account"

publisher = pubsub_v1.PublisherClient()
topic_path = publisher.topic_path(project_id, topic_id)

request = Topic(
    name=topic_path,
    ingestion_data_source_settings=IngestionDataSourceSettings(
        aws_kinesis=IngestionDataSourceSettings.AwsKinesis(
            stream_arn=stream_arn,
            consumer_arn=consumer_arn,
            aws_role_arn=aws_role_arn,
            gcp_service_account=gcp_service_account,
        )
    ),
)

topic = publisher.create_topic(request=request)

print(f"Created topic: {topic.name} with AWS Kinesis Ingestion Settings")

namespace pubsub = ::google::cloud::pubsub;
namespace pubsub_admin = ::google::cloud::pubsub_admin;
[](pubsub_admin::TopicAdminClient client, std::string project_id,
   std::string topic_id, std::string stream_arn, std::string consumer_arn,
   std::string aws_role_arn, std::string gcp_service_account) {
  google::pubsub::v1::Topic request;
  request.set_name(
      pubsub::Topic(std::move(project_id), std::move(topic_id)).FullName());
  auto* aws_kinesis =
      request.mutable_ingestion_data_source_settings()->mutable_aws_kinesis();
  aws_kinesis->set_stream_arn(stream_arn);
  aws_kinesis->set_consumer_arn(consumer_arn);
  aws_kinesis->set_aws_role_arn(aws_role_arn);
  aws_kinesis->set_gcp_service_account(gcp_service_account);

  auto topic = client.CreateTopic(request);
  // Note that kAlreadyExists is a possible error when the library retries.
  if (topic.status().code() == google::cloud::StatusCode::kAlreadyExists) {
    std::cout << "The topic already exists\n";
    return;
  }
  if (!topic) throw std::move(topic).status();

  std::cout << "The topic was successfully created: " << topic->DebugString()
            << "\n";
}

/**
 * TODO(developer): Uncomment these variables before running the sample.
 */
// const topicNameOrId = 'YOUR_TOPIC_NAME_OR_ID';
// const roleArn = 'arn:aws:iam:...';
// const gcpServiceAccount = 'ingestion-account@...';
// const streamArn = 'arn:aws:kinesis:...';
// const consumerArn = 'arn:aws:kinesis:...';

// Imports the Google Cloud client library
import {PubSub} from '@google-cloud/pubsub';

// Creates a client; cache this for further use
const pubSubClient = new PubSub();

async function createTopicWithKinesisIngestion(
  topicNameOrId: string,
  awsRoleArn: string,
  gcpServiceAccount: string,
  streamArn: string,
  consumerArn: string
) {
  // Creates a new topic with Kinesis ingestion.
  await pubSubClient.createTopic({
    name: topicNameOrId,
    ingestionDataSourceSettings: {
      awsKinesis: {
        awsRoleArn,
        gcpServiceAccount,
        streamArn,
        consumerArn,
      },
    },
  });
  console.log(`Topic ${topicNameOrId} created with AWS Kinesis ingestion.`);
}

Pour en savoir plus sur les ARN, consultez les pages Noms de ressource Amazon (ARN) et Identifiants IAM.

Si vous rencontrez des problèmes, consultez Résoudre les problèmes liés à un sujet d'importation Amazon Kinesis Data Streams.

Modifier un sujet d'importation Amazon Kinesis Data Streams

Vous pouvez modifier les paramètres de la source de données d'ingestion d'un sujet d'importation Amazon Kinesis Data Streams. Procédez comme suit :

Quotas et limites pour les sujets d'importation Amazon Kinesis Data Streams

Le débit de l'éditeur pour les sujets d'importation est limité par le quota de publication du sujet. Pour en savoir plus, consultez la page Quotas et limites de Pub/Sub.

Étape suivante

• Surveiller un sujet d'importation

• Choisissez le type d'abonnement pour votre sujet.

• Découvrez comment publier un message dans un sujet.

• Créez ou modifiez un sujet avec la gcloud CLI, les API REST ou les bibliothèques clientes.

• Dépannage d'un sujet d'importation Amazon Kinesis Data Streams

• Procédure de configuration d'une tâche de transfert de stockage pour transférer des données d'Amazon S3 vers GCS