Как подключить модуль в AKS к Центру событий Azure с помощью интерфейса Kafka?

Как подключить потребителя Kafka, написанного на Python, к Центру событий в модуле AKS? Я уже пробовал использовать Workload Identity с сервисным соединителем (ранее я безуспешно пытался использовать строку подключения), но мне все равно не удалось подключиться. Я убедился, что созданное удостоверение имеет необходимые права в Event Hub.

Потребитель был протестирован локально и работает нормально, вот код

from azure.identity import DefaultAzureCredential, WorkloadIdentityCredential, ManagedIdentityCredential
from azure.storage.blob import BlobServiceClient
from confluent_kafka import Consumer, KafkaException
import sys
import getopt
import json
import logging
from functools import partial
from pprint import pformat
import os


def stats_cb(stats_json_str):
    stats_json = json.loads(stats_json_str)
    print('\nKAFKA Stats: {}\n'.format(pformat(stats_json)))


def oauth_cb(cred, namespace_fqdn, config):
    # confluent_kafka requires an oauth callback function to return (str, float) with the values of (<access token>, <expiration date in seconds from epoch>)

    # cred: an Azure identity library credential object. Ex: an instance of DefaultAzureCredential, ManagedIdentityCredential, etc
    # namespace_fqdn: the FQDN for the target Event Hubs namespace. Ex: 'mynamespace.servicebus.windows.net'
    # config: confluent_kafka passes the value of 'sasl.oauthbearer.config' as the config param
    access_token = cred.get_token('https://<eventhubs-namespace>.servicebus.windows.net/.default')
    return access_token.token, access_token.expires_on


def print_usage_and_exit(program_name):
    sys.stderr.write(
        'Usage: %s [options..] <eventhubs-namespace> <group> <topic1> <topic2> ..\n' % program_name)
    options = '''
 Options:
  -T <intvl>   Enable client statistics at specified interval (ms)
'''
    sys.stderr.write(options)
    sys.exit(1)


if __name__ == '__main__':
    optlist, argv = getopt.getopt("<resource group", 'T:')
    if len(argv) < 3:
        print_usage_and_exit(sys.argv[0]) 

      # Azure credential     # See https://docs.microsoft.com/en-us/azure/developer/python/sdk/authentication-overview

    cred = DefaultAzureCredential(managed_identity_client_id = "<id>")


    # Consumer configuration
    # See https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md
    conf = {
        'bootstrap.servers': '<eventhubs-namespace>.servicebus.windows.net:9093',
        'group.id': '$Default',
        'session.timeout.ms': 6000,
        'auto.offset.reset': 'earliest',

        # Required OAuth2 configuration properties
        'security.protocol': 'SASL_SSL',
        'sasl.mechanism': 'OAUTHBEARER',
        # the resulting oauth_cb must accept a single `config` parameter, so we use partial to bind the namespace/identity to our function
        'oauth_cb': partial(oauth_cb, cred, '<eventhubs-namespace>.servicebus.windows.net:9093'),
    }
    #print(str(conf))
    # Check to see if -T option exists
    for opt in optlist:
        if opt[0] != '-T':
            continue
        try:
            intval = int(opt[1])
        except ValueError:
            sys.stderr.write("Invalid option value for -T: %s\n" % opt[1])
            sys.exit(1)

        if intval <= 0:
            sys.stderr.write("-T option value needs to be larger than zero: %s\n" % opt[1])
            sys.exit(1)

        conf['stats_cb'] = stats_cb
        conf['statistics.interval.ms'] = int(opt[1])

    # Create logger for consumer (logs will be emitted when poll() is called)
    logger = logging.getLogger('consumer')
    logger.setLevel(logging.DEBUG)
    handler = logging.StreamHandler()
    handler.setFormatter(logging.Formatter('%(asctime)-15s %(levelname)-8s %(message)s'))
    logger.addHandler(handler)

    # Create Consumer instance
    # Hint: try debug='fetch' to generate some log messages
    c = Consumer(conf, logger=logger)
    #print(str(c.list_topics().topics))
    def print_assignment(consumer, partitions):
        print('Assignment:', partitions)

    # Subscribe to topics
    c.subscribe(["test"], on_assign=print_assignment)

    # Read messages from Kafka, print to stdout
    try:
        while True:
            msg = c.poll(timeout=1.0)
            if msg is None:
                continue
            if msg.error():
                raise KafkaException(msg.error())
            else:
                # Proper message
                sys.stderr.write('%% %s [%d] at offset %d with key %s:\n' %
                                 (msg.topic(), msg.partition(), msg.offset(),
                                  str(msg.key())))
                print(msg.value())

    except KeyboardInterrupt:
        sys.stderr.write('%% Aborted by user\n')

    finally:
        # Close down consumer to commit final offsets.
        c.close()

Вот развертывание на K8s

apiVersion: apps/v1
kind: Deployment
metadata:
  name: consumer
  namespace: default
  labels:
    app: consumer
    azure.workload.identity/use: "true"
spec:
  replicas: 1
  selector:
    matchLabels:
      app: consumer
  template:
    metadata:
      labels:
        app: consumer
    spec:
      serviceAccountName: sc-account-<name>
      containers:       
      - name: consumer         
        image: <image>
        command: ["/venv/bin/python","/consumer.py"]
        envFrom:
          - secretRef:
              name: sc-eventhub<secret>
Почему в Python есть оператор "pass"?
Почему в Python есть оператор "pass"?
Оператор pass в Python - это простая концепция, которую могут быстро освоить даже новички без опыта программирования.
Некоторые методы, о которых вы не знали, что они существуют в Python
Некоторые методы, о которых вы не знали, что они существуют в Python
Python - самый известный и самый простой в изучении язык в наши дни. Имея широкий спектр применения в области машинного обучения, Data Science,...
Основы Python Часть I
Основы Python Часть I
Вы когда-нибудь задумывались, почему в программах на Python вы видите приведенный ниже код?
LeetCode - 1579. Удаление максимального числа ребер для сохранения полной проходимости графа
LeetCode - 1579. Удаление максимального числа ребер для сохранения полной проходимости графа
Алиса и Боб имеют неориентированный граф из n узлов и трех типов ребер:
Оптимизация кода с помощью тернарного оператора Python
Оптимизация кода с помощью тернарного оператора Python
И последнее, что мы хотели бы показать вам, прежде чем двигаться дальше, это
Советы по эффективной веб-разработке с помощью Python
Советы по эффективной веб-разработке с помощью Python
Как веб-разработчик, Python может стать мощным инструментом для создания эффективных и масштабируемых веб-приложений.
0
0
92
1
Перейти к ответу Данный вопрос помечен как решенный

Ответы 1

Ответ принят как подходящий

Выполните следующие действия для проверки подлинности и настройки концентратора событий Azure с помощью потребителя Kafka в AKS на Python.

Сначала настройте Azure Identity, чтобы получить необходимые разрешения для доступа к вашему концентратору событий. Используйте DefaultAzureCredential для обработки аутентификации.

Создайте Dockerfile для контейнеризации вашего приложения Python.

FROM python:3.9-slim

WORKDIR /app

COPY requirements.txt requirements.txt
RUN pip install -r requirements.txt

COPY consumer.py consumer.py

CMD ["python", "consumer.py", "<eventhubs-namespace>", "<group>", "<topic>"]

Создайте и отправьте образ Docker в реестр контейнеров. см. это для настройки Kafka Connect для событий Hubs

docker build -t <your-registry>/kafka-consumer .
docker push <your-registry>/kafka-consumer

Создайте YAML-файл развертывания Kubernetes для развертывания контейнерного приложения .


apiVersion: apps/v1
kind: Deployment
metadata:
  name: kafka-consumer
  namespace: default
  labels:
    app: kafka-consumer
spec:
  replicas: 1
  selector:
    matchLabels:
      app: kafka-consumer
  template:
    metadata:
      labels:
        app: kafka-consumer
    spec:
      serviceAccountName: <your-service-account>
      containers:
        - name: kafka-consumer
          image: <your-registry>/kafka-consumer
          env:
            - name: AZURE_CLIENT_ID
              value: <your-client-id>
            - name: AZURE_TENANT_ID
              value: <your-tenant-id>
            - name: AZURE_CLIENT_SECRET
              value: <your-client-secret>
          resources:
            limits:
              memory: "512Mi"
              cpu: "500m"
            requests:
              memory: "256Mi"
              cpu: "250m"

Примените развертывание к кластеру AKS.

kubectl apply -f kafka-consumer-deployment.yaml

Проверьте журналы, чтобы убедиться, что потребитель работает правильно и подключается к Event Hub.

kubectl logs -f deployment/kafka-consumer

Другие вопросы по теме

Перемещение электронных писем в удаленные элементы с помощью Microsoft Graph API
Предоставление согласия администратора на разрешение API регистрации приложения
Периодические проблемы с триггерами в сетке событий для загрузки и удаления больших двоичных объектов в функции Azure
Ошибка выдачи конвейера Azure Devops при запуске сценария PowerShell. В строке отсутствует терминатор: "
Azure b2c все еще проходит проверку подлинности после выхода из системы
Как условно ограничить доступ к конечной точке API на основе идентификатора аудитории?
Azure B2C/WPF — обработка неудачного потока проверки MFA
Как прочитать исходный PDF-файл в источнике данных индексатора в пользовательском WebApiSkill после включения «Разрешить набору навыков читать данные файла»
Создание таблицы CSV на основе выходных данных Crowdstrike API
Конвейер Azure DevOps Terraform: проблема аутентификации с федеративным и управляемым удостоверением