#!/usr/bin/env python
# Просто виводить стандартний вивід і очікує протягом 10 секунд.
import sys
import time
print("Processing " + sys.stdin.readlines()[0])
time.sleep(10)
Вам треба мати кластер Kubernetes, а також інструмент командного рядка kubectl має бути налаштований для роботи з вашим кластером. Рекомендується виконувати ці настанови у кластері, що має щонайменше два вузли, які не виконують роль вузлів управління. Якщо у вас немає кластера, ви можете створити його, за допомогою minikube або використовувати одну з цих пісочниць:
Вам знадобиться реєстр образів контейнерів, куди ви можете завантажувати образи для запуску у своєму кластері. Цей приклад завдання також передбачає, що у вас вже встановлений Docker локально. ## Запуск служби черги повідомлень {#start-message-queue-service} У цьому прикладі використовується RabbitMQ, проте ви можете адаптувати приклад для використання іншої служби повідомлень типу AMQP. На практиці ви можете налаштувати службу черги повідомлень один раз у кластері та використовувати її для багатьох завдань, а також для служб з тривалим виконанням. Запустіть RabbitMQ таким чином: ```shell # Створіть Service для використання StatefulSet kubectl create -f https://kubernetes.io/examples/application/job/rabbitmq/rabbitmq-service.yaml ``` ```none service "rabbitmq-service" created ``` ```shell kubectl create -f https://kubernetes.io/examples/application/job/rabbitmq/rabbitmq-statefulset.yaml ``` ```none statefulset "rabbitmq" created ``` ## Тестування служби черги повідомлень {#test-the-message-queue-service} Тепер ми можемо спробувати доступ до черги повідомлень. Ми створимо тимчасовий інтерактивний Pod, встановимо деякі інструменти на ньому, і спробуємо працювати з чергами. Спочатку створіть тимчасовий інтерактивний Pod. ```shell # Створіть тимчасовий інтерактивний контейнер kubectl run -i --tty temp --image ubuntu:22.04 ``` ```none Waiting for pod default/temp-loe07 to be running, status is Pending, pod ready: false ... [ previous line repeats several times .. hit return when it stops ] ... ``` Зверніть увагу, що ваше імʼя Pod та запрошення командного рядка будуть відрізнятися. Далі встановіть `amqp-tools`, щоб працювати з чергами повідомлень. Наступні команди показують, що вам потрібно запустити в середовищі інтерактивної оболонки у цьому Pod: ```shell apt-get update && apt-get install -y curl ca-certificates amqp-tools python3 dnsutils ``` Пізніше ви створите образ контейнера, який включає ці пакети. Далі перевірте, що ви можете виявити Service для RabbitMQ: ```shell # Виконайте ці команди всередині Pod # Зауважте, що для служби rabbitmq-service існує DNS-імʼя, яке надається Kubernetes: nslookup rabbitmq-service ``` ```none Server: 10.0.0.10 Address: 10.0.0.10#53 Name: rabbitmq-service.default.svc.cluster.local Address: 10.0.147.152 ``` (адреси IP будуть відрізнятися) Якщо надбудова kube-dns не налаштована правильно, попередній крок може не працювати для вас. Ви також можете знайти IP-адресу для цього Service у змінній середовища: ```shell # виконайте цю перевірку всередині Pod env | grep RABBITMQ_SERVICE | grep HOST ``` ```none RABBITMQ_SERVICE_SERVICE_HOST=10.0.147.152 ``` (IP-адреса буде відрізнятися) Далі перевірте, що ви можете створити чергу, і публікувати та отримувати повідомлення. ```shell # Виконайте ці команди всередині Pod # У наступному рядку rabbitmq-service - це імʼя хоста, за яким можна звертатися до rabbitmq-service. # 5672 - це стандартний порт для rabbitmq. export BROKER_URL=amqp://guest:guest@rabbitmq-service:5672 # Якщо ви не могли отримати доступ до "rabbitmq-service" на попередньому кроці, # використовуйте цю команду натомість: BROKER_URL=amqp://guest:guest@$RABBITMQ_SERVICE_SERVICE_HOST:5672 # Тепер створіть чергу: /usr/bin/amqp-declare-queue --url=$BROKER_URL -q foo -d ``` ```none foo ``` Опублікуйте одне повідомлення в черзі: ```shell /usr/bin/amqp-publish --url=$BROKER_URL -r foo -p -b Hello # І отримайте його назад. /usr/bin/amqp-consume --url=$BROKER_URL -q foo -c 1 cat && echo 1>&2 ``` ```none Hello ``` У останній команді інструмент `amqp-consume` взяв одне повідомлення (`-c 1`) з черги, і передав це повідомлення на стандартний ввід довільної команди. У цьому випадку програма `cat` виводить символи, зчитані зі стандартного вводу, а echo додає символ нового рядка, щоб приклад був читабельним. ## Заповнення черги завданнями {#fill-the-queue-with-tasks} Тепер заповніть чергу деякими симульованими завданнями. У цьому прикладі завдання — це рядки, які потрібно надрукувати. На практиці вміст повідомлень може бути таким: - назви файлів, які потрібно обробити - додаткові прапорці для програми - діапазони ключів у таблиці бази даних - параметри конфігурації для симуляції - номери кадрів сцени для рендерингу Якщо є великі дані, які потрібно лише для читання всіма Podʼами Job, зазвичай це розміщують на спільній файловій системі, наприклад NFS, і монтується це на всі Podʼи тільки для читання, або напишіть програму в Pod так, щоб вона могла нативно читати дані з кластерної файлової системи (наприклад: HDFS). У цьому прикладі ви створите чергу та заповните її, використовуючи інструменти командного рядка AMQP. На практиці ви можете написати програму для заповнення черги, використовуючи бібліотеку клієнтів AMQP. ```shell # Виконайте це на вашому компʼютері, а не в Pod /usr/bin/amqp-declare-queue --url=$BROKER_URL -q job1 -d ``` ```none job1 ``` Додайте елементи до черги: ```shell for f in apple banana cherry date fig grape lemon melon do /usr/bin/amqp-publish --url=$BROKER_URL -r job1 -p -b $f done ``` Ви додали 8 повідомлень у чергу. ## Створення образу контейнера {#create-a-container-image} Тепер ви готові створити образ, який ви будете запускати як Job. Job буде використовувати утиліту `amqp-consume` для читання повідомлення з черги та виконання реальної роботи. Ось дуже простий приклад програми:#!/usr/bin/env python
# Просто виводить стандартний вивід і очікує протягом 10 секунд.
import sys
import time
print("Processing " + sys.stdin.readlines()[0])
time.sleep(10)
apiVersion: batch/v1
kind: Job
metadata:
name: job-wq-1
spec:
completions: 8
parallelism: 2
template:
metadata:
name: job-wq-1
spec:
containers:
- name: c
image: gcr.io/<project>/job-wq-1
env:
- name: BROKER_URL
value: amqp://guest:guest@rabbitmq-service:5672
- name: QUEUE
value: job1
restartPolicy: OnFailure