حالت تاریک
فصل ۹ — مدیریت: ابزارها و لاگگیری
این فصل شامل:
- اسکریپتنویسی و گزینههای administration client
- بررسی REST APIها، ابزارها و utilityها
- مدیریت logهای Kafka و ZooKeeper
- یافتن metricهای JMX
- advertised listenerها و clientها
- tracing با interceptor و header
در فصل ۶ زمان زیادی صرف بحث عمیق دربارهٔ brokerها کردیم و در فصلهای قبل نگرانیهای client را در سراسر کتاب بررسی کردیم. برخی شیوههای توسعه را دیدیم که در بیشتر موقعیتها قابلاعمالاند، اما همیشه محیطهایی وجود دارند که به رسیدگی ویژه نیاز دارند. بهترین راه برای حفظ حرکت یک cluster این است که جریان دادهای که از آن عبور میکند را درک کنیم و فعالیت را در زمان اجرا monitor کنیم. اگرچه بهرهبرداری از Apache Kafka لزوماً همان نوشتن و اجرای برنامههای Java نیست، همچنان نیاز به monitor کردن فایلهای log و آگاهی از آنچه با workloadهایمان میگذرد دارد.
۹.۱ Administration clientها
تا اینجا بیشتر فعالیتهای مدیریت cluster را با command line toolهای همراه Kafka انجام دادهایم. بهطور کلی باید با محیط shell برای راهاندازی و نصب Kafka راحت باشیم. با این حال، گزینههای مفیدی وجود دارد که میتوانیم از این اسکریپتهای آماده فراتر برویم.
۹.۱.۱ Administration در code با AdminClient
یک ابزار مفید برای بررسی، کلاس AdminClient است [1]. اگرچه اسکریپتهای shell Kafka برای دسترسی سریع یا taskهای یکباره عالیاند، موقعیتهایی مانند automation وجود دارند که Java AdminClient واقعاً میدرخشد. AdminClient در همان kafka-clients.jarای است که برای producer و consumer clientها استفاده کردیم. میتوان آن را به پروژهٔ Maven اضافه کرد (به pom.xml فصل ۲ مراجعه کنید) یا در دایرکتوری share/ یا libs/ نصب Kafka یافت.
بیایید ببینیم چگونه میتوانیم دستوری را که قبلاً برای ایجاد topic جدید استفاده کردیم، این بار با AdminClient اجرا کنیم. لیستینگ ۹.۱ نشان میدهد چگونه این کار را در فصل ۲ از command line انجام دادیم.
لیستینگ ۹.۱ — ایجاد topic kinaction_selfserviceTopic از command line
bash
bin/kafka-topics.sh # Uses the kafka-topic.sh
--create --topic kinaction_selfserviceTopic \ # script to create a new topic
--bootstrap-server localhost:9094 \
--partitions 2 \ # Includes our custom integers for the number
--replication-factor 2 # of partitions and replicas for our topicاگرچه این مثال command line خوب کار میکند، نمیخواهیم هر بار که کسی به topic جدید نیاز دارد این دستور فراخوانی شود. در عوض، portal self-service میسازیم که developerهای دیگر بتوانند topic جدید روی cluster توسعهٔ ما ایجاد کنند. فرم برنامهٔ ما نام topic و اعداد مربوط به پارتیشنها و replicaها را میگیرد. شکل ۹.۱ نمونهای از نحوهٔ راهاندازی این برنامه برای کاربر نهایی را نشان میدهد. پس از submit فرم وب، کد Java AdminClient اجرا میشود و topic جدید ایجاد میکند.
در این مثال میتوانیم logic اضافه کنیم تا naming convention برای topicهای جدید الگوی مشخصی را رعایت کند (اگر چنین نیازمندی کسبوکاری داشته باشیم). این راهی است برای کنترل بیشتر cluster بهجای اینکه کاربران از command line toolها کار کنند. برای شروع باید کلاس NewTopic بسازیم. constructor این کلاس سه آرگومان میگیرد:
- نام topic
- تعداد پارتیشنها
- تعداد replicaها
شکل ۹.۱. برنامهٔ وب self-service Kafka
Browser Application server Custom application uses AdminClient and NewTopic Self service Kafka cluster New topic request: Name: Custom Create topics Partitions: application Replicas: Submit NewTopic End user submits New topic a web form kinaction_selfserviceTopic created
پس از داشتن این اطلاعات، میتوانیم از شیء AdminClient برای تکمیل کار استفاده کنیم. AdminClient یک شیء Properties میگیرد که همان propertyهایی را دارد که با clientهای دیگر استفاده کردیم، مانند bootstrap.servers و client.id. توجه کنید کلاس AdminClientConfig (http://mng.bz/8065) constantهایی برای مقادیر پیکربندی مانند BOOTSTRAP_SERVERS_CONFIG بهعنوان کمک برای این نامها نگه میدارد. سپس متد createTopics را روی client فراخوانی میکنیم. توجه کنید نتیجه، topicResult، یک شیء Future است. لیستینگ ۹.۲ نشان میدهد چگونه از کلاس AdminClient برای ایجاد topic جدید به نام kinaction_selfserviceTopic استفاده کنیم.
لیستینگ ۹.۲ — استفاده از AdminClient برای ایجاد topic
java
NewTopic requestedTopic =
new NewTopic("kinaction_selfserviceTopic", 2,(short) 2);
AdminClient client =
AdminClient.create(kaProperties);
CreateTopicsResult topicResult =
client.createTopics(
List.of(requestedTopic));
topicResult.values().
get("kinaction_selfserviceTopic").get();در حال حاضر API synchronous نیست، اما میتوانیم با استفاده از تابع get() فراخوانی synchronous بسازیم. در مورد ما یعنی از متغیر topicResult شروع کنیم و شیء Future برگرداندهشده برای topic مشخص را ارزیابی کنیم.
از آنجا که این API هنوز در حال تکامل است، فهرست زیر از taskهای administrative که با AdminClient قابل انجاماند فقط چند function رایج موجود در زمان نگارش را برجسته میکند [1]:
- تغییر configurationها
- ایجاد/حذف/فهرست access control listها (ACL)
- ایجاد پارتیشن
- ایجاد/حذف/فهرست topicها
- describe/list کردن consumer groupها
- describe کردن clusterها
AdminClient ابزار عالی برای ساخت برنامهٔ user-facing برای کسانی است که معمولاً نیازی به اسکریپتهای shell Kafka ندارند یا نمیخواهند از آنها استفاده کنند. همچنین راهی برای کنترل و monitor کردن آنچه روی cluster انجام میشود فراهم میکند.
۹.۱.۲ kcat
kcat (https://github.com/edenhill/kcat) ابزار handy برای داشتن روی workstation است، بهویژه هنگام اتصال remote به clusterها. در حال حاضر تمرکز آن بر producer و consumer client بودن است که metadata cluster را هم میدهد. اگر بخواهید سریع با topic کار کنید و کل toolset Kafka روی ماشین فعلی دانلود نشده، این executable کمک میکند نیازی به shell یا bat scriptها نداشته باشید.
لیستینگ ۹.۳ نشان میدهد چگونه سریع داده را به topic بفرستیم با kcat [2]. با kafka-console-producer فصل ۲ مقایسه کنید.
لیستینگ ۹.۳ — استفاده از kcat بهعنوان producer
bash
# Sends a broker and topic
kcat -P -b localhost:9094 \ # name from our cluster to
-t kinaction_selfserviceTopic # write messages to that topic
// vs. the shell script we used before
bin/kafka-console-producer.sh --bootstrap-server localhost:9094 \
--topic kinaction_selfserviceTopic
# A reminder of the same
# functionality as the console
# producer commandدر لیستینگ ۹.۳ توجه کنید آرگومان -P به kcat داده میشود تا producer mode فعال شود و بتوانیم پیام به cluster بفرستیم. از flag -b برای broker list و -t برای نام topic هدف استفاده میکنیم. چون ممکن است بخواهیم مصرف این پیامها را هم تست کنیم، بیایید ببینیم چگونه kcat را بهعنوان consumer استفاده کنیم (لیستینگ ۹.۴). مانند قبل، لیستینگ ۹.۴ مقایسهٔ kcat با kafka-console-consumer را نشان میدهد. توجه کنید اگرچه flag -C consumer mode را فعال میکند، اطلاعات broker با همان پارامتر producer mode ارسال میشود [2].
لیستینگ ۹.۴ — استفاده از kcat بهعنوان consumer
bash
# Sends a broker and topic
kcat -C -b localhost:9094 \ # name from our cluster to read
-t kinaction_selfserviceTopic # messages from that topic
// vs. the shell script we used before
bin/kafka-console-consumer.sh --bootstrap-server localhost:9094 \
--topic kinaction_selfserviceTopic
# A reminder of the same
# functionality as the console
# consumer commandداشتن راه سریع برای تست topicها و جمعآوری metadata cluster این utility کوچک را در toolbox ما ارزشمند میکند. اما در این مرحله ممکن است بپرسید آیا ابزار دیگری غیر از command line وجود دارد. خبر خوب این است که بله! برای علاقهمندان به REST، Confluent REST Proxy وجود دارد.
۹.۱.۳ Confluent REST Proxy API
گاهی کاربران cluster ترجیح میدهند از APIهای RESTful استفاده کنند چون راه رایجی برای کار بین applicationهاست، چه بهخاطر ترجیح و چه سهولت استفاده. همچنین شرکتهایی با قوانین سخت firewall دربارهٔ portها ممکن است در باز کردن portهایی مانند آنچه تا اینجا برای اتصال broker استفاده کردیم (مثلاً 9094) محتاط باشند [3]. یک گزینهٔ خوب استفاده از Confluent REST Proxy API است (شکل ۹.۲). این proxy یک application جداست که احتمالاً برای production روی سرور خودش host میشود و functionality آن شبیه utility kcat است.
شکل ۹.۲. Confluent REST Proxy topicها را lookup میکند
Kafka REST proxy server Kafka cluster HTTP request Kafka protocol End user REST call JSON response Topic list returned: to client kinaction_alerttrend
در زمان نگارش، functionهای administration محدود به query کردن state cluster است. مستندات Confluent گزینههای administration را بهعنوان featureهای پشتیبانیشده در آینده فهرست میکند [4]. برای استفاده و تست REST Proxy، بیایید آن را طبق لیستینگ ۹.۵ راهاندازی کنیم. برای کار، باید ZooKeeper و Kafka از قبل در حال اجرا باشند.
لیستینگ ۹.۵ — راهاندازی REST Proxy
bash
bin/kafka-rest-start.sh \ # Run this command from the installed
etc/kafka-rest/kafka-rest.properties # Kafka folder to start the REST endpoint.چون با فهرست topicها آشنا هستیم، بیایید ببینیم چگونه با REST Proxy و دستوری مانند curl به HTTP endpoint بزنیم [5]. چون GET request است، میتوانیم http://localhost:8082/topics را در browser کپی کنیم و نتیجه را ببینیم.
لیستینگ ۹.۶ — فراخوانی cURL به REST Proxy برای فهرست topic
bash
curl -X GET \
-H "Accept: application/vnd.kafka.v2+json" \
# Specifies a format
localhost:8082/topics
# Our target, the endpoint /topics, lists the # and version
# topics we've created and Kafka's internal topics.
// Output:
["__confluent.support.metrics","_confluent-metrics",
"_schemas","kinaction_alert"] # Sample output of
# the curl commandاستفاده از ابزاری مانند curl به ما اجازه میدهد header ارسالی را کنترل کنیم. Accept در لیستینگ ۹.۶ به cluster میگوید چه format و نسخهای استفاده میکنیم — v2 بهعنوان نسخهٔ API و format JSON مربوط به metadata requestها.
یادداشت چون این API در حال تکامل است، با «Confluent REST Proxy API Reference» در http://mng.bz/q5Nw همگام بمانید تا نسخههای جدیدتر با featureهای بیشتر منتشر شوند.
۹.۲ اجرای Kafka بهعنوان سرویس systemd
یک تصمیم دربارهٔ اجرای Kafka این است که broker start و restart چگونه انجام شود. کسانی که با مدیریت سرورها بهعنوان سرویس Linux با ابزاری مانند Puppet (https://puppet.com/) آشنا هستند، احتمالاً با نصب service unit file آشنااند و میتوانند از آن دانش برای instanceهای در حال اجرا با systemd استفاده کنند. برای کسانی که با systemd آشنا نیستند: systemd componentها را در سراسر سیستم initialize و نگه میدارد [6]. یک روش رایج تعریف ZooKeeper و Kafka بهصورت unit file برای systemd است.
لیستینگ ۹.۷ بخشی از نمونه service unit file را نشان میدهد که سرویس ZooKeeper را هنگام start سرور راهاندازی میکند. همچنین پس از خروج غیرعادی ZooKeeper را restart میکند. در عمل یعنی چیزی مانند kill -9 روی process ID (PID) که restart فرآیند را trigger میکند. اگر Confluent tar را در setup نصب کردید (در صورت نیاز به appendix A مراجعه کنید)، فایل service نمونه در مسیر lib/systemd/system/confluent-zookeeper.service است. مستندات «Using Confluent Platform systemd Service Unit Files» در http://mng.bz/7lG9 جزئیات استفاده از این فایلها را میدهد. unit file در لیستینگ باید شبیه نحوهٔ start ZooKeeper در مثالهای ما آشنا به نظر برسد.
لیستینگ ۹.۷ — unit file ZooKeeper
ini
... # Captures the start command to run
[Service] # ZooKeeper (similar to what we
... # manually ran to start ZooKeeper)
ExecStart=/opt/kafkainaction/bin/zookeeper-server-start.sh
/opt/kafkainaction/config/zookeeper.properties
ExecStop=
/opt/kafkainaction/bin/zookeeper-server-stop.sh # Shuts down the
Restart=on-abnormal # Runs ExecStart if an error # ZooKeeper instance
... # condition causes a failureفایل نمونه برای سرویس Kafka هم در Confluent tar در مسیر lib/systemd/system/confluent-kafka.service وجود دارد. لیستینگ بعدی نشان میدهد چون unit fileها تعریف شدهاند، میتوانیم سرویسها را با دستورات systemctl مدیریت کنیم [6].
لیستینگ ۹.۸ — راهاندازی Kafka با systemctl
bash
sudo systemctl start zookeeper # Starts the
sudo systemctl start kafka # Starts the # ZooKeeper service
# Kafka serviceاگر از فایلهای نمونه bundle Confluent استفاده میکنید، پس از unzip پوشهٔ root، داخل ../lib/systemd/system را برای نمونه service fileهای دیگر ببینید. برخی شامل Connect، Schema Registry و REST API هستند.
۹.۳ لاگگیری
علاوه بر event logهای Kafka که دادهٔ event ما را نگه میدارند، موارد دیگری هست که باید به یاد داشته باشیم: logهای application که Kafka بهعنوان برنامهٔ در حال اجرا تولید میکند. logهای این بخش event و پیام از سرورهای Kafka نیستند، بلکه خروجی بهرهبرداری خود Kafka است. ZooKeeper را هم نباید فراموش کنیم!
۹.۳.۱ logهای application Kafka
اگرچه ممکن است به یک فایل log برای کل application عادت داشته باشیم، Kafka چند فایل log دارد که برای troubleshooting ممکن است به آنها نیاز داشته باشیم. بهخاطر چند فایل، شاید مجبور شویم appenderهای Log4j مختلف را تغییر دهیم تا view لازم از عملیات را حفظ کنیم.
کدام Kafka appender؟
kafkaAppenderهمانKafkaLog4jAppenderنیست (http://mng.bz/5ZpB). برای استفاده ازKafkaLog4jAppenderبهعنوان appender باید خط زیر را بهروزرسانی کنیم و dependency برای client و appender JARهای همان نسخه را اضافه کنیم، بهجای مقدارorg.apache.log4j.ConsoleAppender:propertieslog4j.appender.kafkaAppender= org.apache.kafka.log4jappender.KafkaLog4jAppenderxml<dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-log4j-appender</artifactId> <version>2.7.1</version> </dependency>این نگاه جالبی به فرستادن مستقیم logها به Kafka است. برخی راهحلها خود فایلهای log را parse میکنند و سپس به Kafka میفرستند.
بهطور پیشفرض، logهای سرور پیوسته به دایرکتوری اضافه میشوند. با این حال هیچ logی حذف نمیشود و اگر این فایلها برای audit یا troubleshooting لازم باشند، ممکن است رفتار ترجیحی باشد. اگر بخواهیم تعداد و اندازه را کنترل کنیم، سادهترین راه بهروزرسانی فایل log4j.properties قبل از start broker است. لیستینگ ۹.۹ دو property مهم برای kafkaAppender تنظیم میکند: MaxFileSize و MaxBackupIndex [7].
لیستینگ ۹.۹ — retention log سرور Kafka
properties
# Defines the file size to determine
log4j.appender.kafkaAppender.MaxFileSize=500KB # when to create a new log file
log4j.appender.kafkaAppender.MaxBackupIndex=10
# Sets the number of older files to keep,
# which helps if we want more than the
# current log for troubleshootingتوجه کنید تغییر kafkaAppender فقط نحوهٔ برخورد با فایل server.log را تغییر میدهد. اگر بخواهیم اندازه و تعداد backup فایل متفاوت برای فایلهای مختلف Kafka اعمال کنیم، میتوانیم از جدول appender به filename استفاده کنیم. در جدول ۹.۱، نام appender در ستون چپ کلید logging است که نحوهٔ ذخیرهٔ log در brokerها را تحت تأثیر قرار میدهد [8].
جدول ۹.۱ — appender به الگوی log
| Appender name | Log filename |
|---|---|
| kafkaAppender | server.log |
| stateChangeAppender | state-change.log |
| requestAppender | kafka-request.log |
| cleanerAppender | log-cleaner.log |
| controllerAppender | controller.log |
| authorizerAppender | kafka-authorizer.log |
تغییر فایل log4j.properties نیاز به restart broker دارد؛ پس بهتر است قبل از اولین start broker، در صورت امکان، نیازمندیهای logging را مشخص کنیم. میتوانیم مقدار را با JMX هم تغییر دهیم، اما مقدار پس از restart broker پایدار نمیماند.
اگرچه در این بخش روی logهای Kafka تمرکز کردیم، باید logهای ZooKeeper را هم بررسی کنیم. چون ZooKeeper مانند brokerها اجرا و log میکند، باید به خروجی logging آن سرورها هم توجه کنیم.
۹.۳.۲ logهای ZooKeeper
بسته به نحوهٔ نصب و مدیریت ZooKeeper، ممکن است نیاز به تغییر پیکربندی logging آن هم داشته باشیم. پیکربندی پیشفرض ZooKeeper فایل log را حذف نمیکند، اما نصب Kafka ما ممکن است این feature را اضافه کرده باشد. اگر setup node محلی ZooKeeper را از appendix A دنبال کردید، این مقادیر در config/zookeeper.properties قابل تنظیماند. در هر صورت خوب است retention logهای application ZooKeeper با مقادیر زیر کنترل شود و مطابق نیاز troubleshooting باشد:
autopurge.purgeInterval— فاصلهٔ زمانی (ساعت) که purge trigger میشود. باید بالای ۰ تنظیم شود تا cleanup انجام شود [9].autopurge.snapRetainCount— تعداد snapshotهای اخیر و transaction logهای مرتبط درdataDirوdataLogDir[9]. پس از عبور از این تعداد، فایلهای log قدیمیتر حذف میشوند. بسته به نیاز، ممکن است بیشتر یا کمتر نگه داریم. مثلاً اگر log فقط برای troubleshooting است، retention کمتر از سناریوی audit لازم است.snapCount— ZooKeeper transactionها را در transaction log مینویسد. این مقدار تعیین میکند چند transaction در یک فایل log شود. اگر با اندازهٔ کل فایلها مشکل داریم، شاید کمتر از پیشفرض (100,000) تنظیم کنیم [10].
راهحلهای دیگری برای log rotation و cleanup وجود دارد؛ مثلاً logrotate ابزار مفیدی برای rotation و فشردهسازی log است.
نگهداری فایل log وظیفهٔ administration مهمی است. با این حال taskهای دیگری هم هنگام rollout cluster Kafka جدید باید در نظر بگیریم. یکی اطمینان از اتصال clientها به brokerهاست.
۹.۴ Firewallها
بسته به پیکربندی شبکه، ممکن است clientهای داخل شبکه یا خارج از شبکهای که brokerهای Kafka در آن هستند سرویس دهیم [3]. brokerهای Kafka میتوانند روی چند port گوش دهند. مثلاً پیشفرض port plain text 9092 است. port SSL روی 9093 هم روی همان host قابل تنظیم است. بسته به نحوهٔ اتصال clientها، ممکن است هر دو port باید باز باشند.
علاوه بر این، ZooKeeper port 2181 را برای اتصال client شامل میشود. port 2888 توسط nodeهای follower ZooKeeper برای اتصال به leader استفاده میشود و port 3888 هم بین nodeهای ZooKeeper برای ارتباط [11]. اگر remote برای JMX یا سرویسهای Kafka دیگر (مانند REST Proxy) وصل میشوید، exposure آن port به محیطها یا کاربران دیگر را در نظر بگیرید. بهطور کلی اگر command line toolی نیاز به port در انتهای hostname برای ZooKeeper یا Kafka دارد، باید این portها قابل دسترس باشند، بهویژه اگر firewall وجود دارد.
۹.۴.۱ Advertised listenerها
یک خطای اتصال که اغلب شبیه مشکل firewall به نظر میرسد، استفاده از propertyهای listeners و advertised.listeners است. clientها باید hostname درست — اگر داده شده — را برای اتصال استفاده کنند؛ پس باید hostname قابل reach باشد، هرچقدر قوانین تنظیم شده باشند. مثلاً بیایید listeners در مقابل advertised.listeners را ببینیم جایی که این مقادیر یکسان نیستند.
تصور کنیم به broker وصل میشویم و هنگام start client اتصال برقرار میشود، اما هنگام consume پیام نه. این رفتار «ناهماهنگ» چگونه ممکن است؟ به یاد داشته باشید وقتی client start میکند، برای metadata دربارهٔ brokerی که باید وصل شود به هر broker متصل میشود. اتصال اولیه از اطلاعات در پیکربندی listeners Kafka استفاده میکند. آنچه برای اتصال بعدی به client برمیگرداند دادهٔ advertised.listeners است [12]. پس محتمل است client برای کار به host دیگری وصل شود.
شکل ۹.۳ نشان میدهد client برای اولین تلاش اتصال از یک hostname و برای دومین اتصال hostname دیگری استفاده میکند. hostname دوم در فراخوانی اولیه به client بهعنوان محل جدید اتصال داده شده است.
تنظیم مهم inter.broker.listener.name است که تعیین میکند brokerها در cluster چگونه به هم وصل شوند [12]. اگر brokerها به هم نرسند، replicaها fail میشوند و cluster در وضعیت خوبی نخواهد بود! برای توضیح عالی advertised listenerها، مقالهٔ Robin Moffatt «Kafka Listeners – Explained» را ببینید [12]. شکل ۹.۳ از diagramهای همان سایت الهام گرفته [12].
شکل ۹.۳. advertised listenerهای Kafka در مقابل listenerها
Scenario 1: no advertised listeners. Producer client starts and requests metadata from bootstrap server. Successful connection Network 1 Network 2 as URL resolves for both networks Kafka cluster Bootstrap server Broker Broker 0 X Client X ???? Connection fails as it can't Returns internal resolve network 2 address. listener name Scenario 2: advertised listeners with URL resolved by both networks. Producer client requests metadata. Network 1 Network 2 Successful connection Kafka cluster Bootstrap server Broker Broker 0 X Client Returns public advertised listener Successful resolvable by both networks. connection
۹.۵ Metricها
در فصل ۶ نمونهای از راهاندازی دیدن metricهای JMX از application را دیدیم. توانایی دیدن آن metricها گام اول است. بیایید نگاهی به یافتن metricهایی بیندازیم که احتمالاً نواحی نگرانکننده را برجسته میکنند.
۹.۵.۱ کنسول JMX
میتوان از GUI برای کاوش metricهای exposeشده استفاده کرد. VisualVM (https://visualvm.github.io/) یک مثال است. نگاه به metricهای JMX موجود کمک میکند نقاط interest را کشف کنیم که شاید alert اضافه کنیم. هنگام نصب VisualVM، حتماً MBeans Browser را هم نصب کنید.
همانطور که در فصل ۶ گفته شد، برای هر broker که میخواهیم وصل شویم باید JMX_PORT تعریف شود. میتوان با environment variable در terminal این کار را کرد:
bash
export JMX_PORT=49999[13] مطمئن شوید برای هر broker و هر node ZooKeeper scope جداگانه درست تنظیم شده.
KAFKA_JMX_OPTS گزینهٔ دیگری برای اتصال remote به brokerهای Kafka است. port و hostname درست را یادداشت کنید. لیستینگ ۹.۱۰ مثالی با آرگومانهای مختلف [13] — port 49999 و localhost بهعنوان hostname — نشان میدهد. پارامترهای دیگر اتصال بدون SSL و بدون authentication را ممکن میکنند.
لیستینگ ۹.۱۰ — گزینههای JMX Kafka
bash
# Sets the hostname for
KAFKA_JMX_OPTS="-Djava.rmi.server.hostname=127.0.0.1 # the localhost RMI server
-Dcom.sun.management.jmxremote.local.only=false
# Allows remote connections
# Exposes -Dcom.sun.management.jmxremote.rmi.port=49999
# this port -Dcom.sun.management.jmxremote.authenticate=false # Turns off authentication
# for JMX -Dcom.sun.management.jmxremote.ssl=false" # and SSL checksبیایید یک metric کلیدی broker و یافتن مقدار مورد نیاز را با کمک شکل ۹.۴ ببینیم که نمایش کوچک MBeans برای مقدار UnderReplicatedPartitions را نشان میدهد. با نامی مانند
kafka.server:type=ReplicaManager,name=UnderReplicatedPartitionsمیتوانیم ساختار شبیه پوشه را از kafka.server drill down کنیم.
شکل ۹.۴. محل
UnderReplicatedPartitionsوRequestQueueSizeWe can drill down to find out MBeans if topic kinaction_alert has underreplicated partitions, for example. kafka.server kafka.network Type Type ReplicaManager RequestChannel UnderReplicatedPartitions RequestQueueSize Value Value 0 0
ادامه میدهیم و type ReplicaManager با attribute UnderReplicatedPartitions را پیدا میکنیم. RequestQueueSize هم در شکل ۹.۴ بهعنوان مثال دیگر یافتن مقدار نشان داده شده [14]. حالا که میدانید چگونه به مقادیر مشخص browse کنید، بیایید مهمترین موارد روی سرورها را بررسی کنیم.
اگر Confluent Control Center یا Confluent Cloud استفاده میکنید، بیشتر این metricها در monitoring داخلی به کار میروند. Confluent Platform پیشنهاد میکند alert روی سه مقدار زیر شروع شود: UnderMinIsrPartitionCount، UnderReplicatedPartitions، UnderMinIsr [14].
در بخش بعد با interceptorها monitoring متفاوتی را بررسی میکنیم.
۹.۶ گزینه tracing
metricهای built-in که تا اینجا دیدیم snapshot خوبی از سلامت فعلی میدهند، اما اگر بخواهیم یک پیام را در سیستم trace کنیم چه؟ چه چیزی produced message و وضعیت consumed آن را نشان میدهد؟ بیایید مدل ساده اما مستقیمی را بررسی کنیم که شاید برای نیازمندیهای ما کار کند.
فرض کنیم producer داریم که هر event ID یکتا دارد. چون هر پیام مهم است، نمیخواهیم eventای از دست برود. با یک client، business logic عادی اجرا میشود و پیامها را از topic مصرف میکند. در این حالت منطقی است ID event پردازششده را در database یا flat file log کنیم. consumer جدا — auditing consumer — از همان topic داده میگیرد و مطمئن میشود IDای از processed entryهای application اول گم نشده. اگرچه این فرآیند خوب کار میکند، logic به application اضافه میکند و شاید بهترین انتخاب نباشد.
شکل ۹.۵ رویکرد متفاوت با Kafka interceptor را نشان میدهد. در عمل، interceptor تعریفشده راهی برای افزودن logic به producer، consumer یا هر دو است با hook به جریان عادی client، intercept کردن record و افزودن دادهٔ سفارشی قبل از ادامهٔ مسیر عادی. تغییرات ما configuration-driven است و logic خاص را تا حد زیادی از client جدا نگه میدارد.
بیایید مفهوم interceptor را که در فصل ۴ بهاختصار دیدیم مرور کنیم — producer interceptor چه میتواند برای پیامها انجام دهد. با افزودن interceptor روی producer و consumer client، میتوانیم logic monitoring را از logic application جدا کنیم. concern crosscutting monitoring امیدواریم با این رویکرد encapsulatedتر شود.
شکل ۹.۵. interceptorها برای tracing
kinactionTraceId kinactionTraceId Producer client added at this step Kafka read at this step Consumer Send OnSend OnConsume Poll Sends initial Adds header trace ID Called before the consumer Consumes your message to each record and poll method. You look for record, header then logs to standard and log your trace ID header and all! out (stdout) to standard out (stdout).
۹.۶.۱ منطق producer
جالب است بدانیم میتوانیم بیش از یک interceptor داشته باشیم؛ پس لازم نیست همهٔ logic در یک class باشد و بعداً interceptor دیگر اضافه یا حذف کنیم. ترتیب فهرست classها مهم است چون همان ترتیب اجرای logic است. اولین interceptor record را از producer client میگیرد. اگر interceptor record را تغییر دهد، interceptorهای بعدی در chain همان record دقیق اول را نمیبینند [15].
بیایید با interface Java ProducerInterceptor شروع کنیم. این interceptor جدید را به Alert producer فصل ۴ اضافه میکنیم. class جدید AlertProducerMetricsInterceptor logic حول alertهای produced را اضافه میکند، مانند لیستینگ ۹.۱۱. implement کردن ProducerInterceptor به lifecycle interceptor producer hook میدهد. logic در متد onSend توسط متد send() از producer client عادی فراخوانی میشود [15]. در لیستینگ header به نام kinactionTraceId اضافه میکنیم. ID یکتا کمک میکند در سمت consumption تأیید کنیم همان پیامی را میبینیم که در ابتدا produce شده.
لیستینگ ۹.۱۱ — مثال AlertProducerMetricsInterceptor
java
public class AlertProducerMetricsInterceptor
implements ProducerInterceptor<Alert, String> {
final static Logger log =
LoggerFactory.getLogger(AlertProducerMetricsInterceptor.class);
public ProducerRecord<Alert, String>
onSend(ProducerRecord<Alert, String> record) {
Headers headers = record.headers();
String kinactionTraceId = UUID.randomUUID().toString();
headers.add("kinactionTraceId",
kinactionTraceId.getBytes());
log.info("kinaction_info Created kinactionTraceId: {}", kinactionTraceId);
return record;
}
public void onAcknowledgement(
RecordMetadata metadata, Exception exception)
{
if (exception != null) {
log.info("kinaction_error " + exception.getMessage());
} else {
log.info("kinaction_info topic = {}, offset = {}",
metadata.topic(), metadata.offset());
}
}
// rest of the code omitted
}باید کلاس موجود AlertProducer را هم تغییر دهیم تا interceptor جدید ثبت شود. property interceptor.classes را به پیکربندی producer با مقدار full package name کلاس جدید اضافه میکنیم: AlertProducerMetricsInterceptor. اگرچه برای وضوح از نام property استفاده کردیم، constant کلاس ProducerConfig را هم میتوان به کار برد: ProducerConfig.INTERCEPTOR_CLASSES_CONFIG [15]. لیستینگ ۹.۱۲ این تغییر لازم را نشان میدهد.
لیستینگ ۹.۱۲ — AlertProducer با پیکربندی interceptor
java
Properties kaProperties = new Properties();
...
kaProperties.put("interceptor.classes",
AlertProducerMetricsInterceptor.class.getName());
Producer<Alert, String> producer =
new KafkaProducer<Alert, String>(kaProperties);در این مثال یک interceptor داریم که ID یکتا برای هر پیام produced log میکند. این ID را بهعنوان header به record اضافه میکنیم تا وقتی consumer پیام را pull میکند، consumer interceptor متناظر ID پردازششده را log کند. هدف monitoring end-to-end خودمان خارج از Kafka است. با parse کردن logهای application، پیامهایی مانند لیستینگ ۹.۱۳ از کلاس AlertProducerMetricsInterceptor میبینیم.
لیستینگ ۹.۱۳ — خروجی interceptor alert
kinaction_info Created kinactionTraceId:
603a8922-9fb5-442a-a1fa-403f2a6a875d
kinaction_info topic = kinaction_alert, offset = 1۹.۶.۲ منطق consumer
حالا که interceptor ارسال پیام را راهاندازی کردیم، باید logic مشابه در سمت consumer پیاده کنیم. میخواهیم تأیید کنیم همان مقدار header که با producer interceptor اضافه کردیم در consumption دیده میشود. لیستینگ ۹.۱۴ implement ConsumerInterceptor برای بازیابی این header [16] را نشان میدهد.
لیستینگ ۹.۱۴ — مثال AlertConsumerMetricsInterceptor
java
public class AlertConsumerMetricsInterceptor
implements ConsumerInterceptor<Alert, String> {
public ConsumerRecords<Alert, String>
onConsume(ConsumerRecords<Alert, String> records) {
if (records.isEmpty()) {
return records;
} else {
for (ConsumerRecord<Alert, String> record : records) {
Headers headers = record.headers();
for (Header header : headers) {
if ("kinactionTraceId".equals(
header.key())) {
log.info("KinactionTraceId is: " + new String(header.value()));
}
}
}
}
return records;
}
}مانند producer، interface مخصوص consumer یعنی ConsumerInterceptor را برای interceptor جدید استفاده کردیم. همهٔ recordها و headerها را loop کردیم تا key سفارشی kinactionTraceId را پیدا و به standard output بفرستیم. کلاس موجود AlertConsumer را هم تغییر دادیم تا interceptor جدید ثبت شود. property interceptor.classes باید به پیکربندی consumer با full package name کلاس جدید اضافه شود: AlertConsumerMetricsInterceptor. لیستینگ ۹.۱۵ این گام لازم را نشان میدهد.
لیستینگ ۹.۱۵ — AlertConsumer با پیکربندی interceptor
java
public class AlertConsumer {
Properties kaProperties = new Properties();
...
kaProperties.put("group.id",
"kinaction_alertinterceptor");
kaProperties.put("interceptor.classes",
AlertConsumerMetricsInterceptor.class.getName());
...
}اگر بیش از یک class لازم است، لیست comma-separated میتواند باشد [16]. constant کلاس ConsumerConfig هم موجود است: ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG [16]. اگرچه interceptor در هر دو انتهای flow را دیدیم، راه دیگر افزودن functionality به client code — override کردن clientها — هم وجود دارد.
۹.۶.۳ Override کردن clientها
اگر source code clientهایی که developerهای دیگر استفاده میکنند را کنترل میکنیم، میتوانیم subclass از client موجود بسازیم یا client خودمان که interface producer/consumer Kafka را implement میکند. در زمان نگارش، پروژهٔ Brave (https://github.com/openzipkin/brave) نمونهای از clientی دارد که با tracing data کار میکند.
برای کسانی که با Brave آشنا نیستند: کتابخانهای برای افزودن instrumentation به distributed tracing است. میتواند داده را به چیزی مانند Zipkin server (https://zipkin.io/) بفرستد که جمعآوری و جستجو را handle میکند. اگر علاقهمندید، کلاس TracingConsumer (http://mng.bz/6mAo) را برای مثال real-world افزودن functionality به client با Kafka ببینید.
میتوانیم producer و consumer client را decorate کنیم تا tracing (یا logic سفارشی) فعال شود، اما در stub زیر روی consumer تمرکز میکنیم. کد لیستینگ ۹.۱۶ بخشی pseudo code برای افزودن logic سفارشی به جریان consumer عادی Kafka است. developerهایی که میخواهند پیام با logic سفارشی مصرف کنند از instance KInActionCustomConsumer استفاده میکنند که reference به consumer عادی به نام normalKafkaConsumer دارد. logic سفارشی رفتار لازم را اضافه میکند در حالی که با client سنتی تعامل دارد. developerها با consumer شما کار میکنند که client عادی را پشت صحنه handle میکند.
لیستینگ ۹.۱۶ — consumer client سفارشی
java
final class KInActionCustomConsumer<K, V> implements Consumer<K, V> {
...
final Consumer<K, V> normalKafkaConsumer;
@Override
public ConsumerRecords<K, V> poll(
final Duration timeout)
{
//Custom logic here
// Normal Kafka consumer used as normal
return normalKafkaConsumer.poll(timeout);
}
...
}این لیستینگ فقط comment نشان میدهد logic شما کجا برود، اما کاربران در صورت تمایل از متدهای client عادی abstract شدهاند در حالی که code سفارشی مانند بررسی duplicate submission یا log tracing data از header اجرا میشود. رفتار اضافه مانع client عادی نمیشود.
۹.۷ ابزارهای monitoring عمومی
چون Kafka یک application Scala™ است، میتواند از JMX و کتابخانهٔ Yammer Metrics استفاده کند [17]. این کتابخانه metricهای JMX روی بخشهای مختلف application را فراهم میکند و گزینههایی را که میتوانیم ارزیابی کنیم دیدیم. با گسترش استفاده از Kafka، ابزارهایی هستند که علاوه بر metricهای JMX از دستورات administration و techniqueهای دیگر برای clusterهای قابل مدیریت استفاده میکنند. البته فهرست زیر کامل نیست و featureهای ابزارها ممکن است تغییر کند. با این حال بیایید چند گزینه را ببینیم.
Cluster Manager for Apache Kafka یا CMAK (https://github.com/yahoo/CMAK)، قبلاً Kafka Manager، پروژهٔ جالبی است که علاوه بر UI برای فعالیتهای administrative مختلف، مدیریت Kafka را هدف میگیرد و از Yahoo™! به اشتراک گذاشته شده. یک feature کلیدی مدیریت چند cluster است. featureهای دیگر شامل inspection state کلی cluster و تولید و اجرای partition reassignment است. این ابزار authentication کاربر با LDAP را هم handle میکند که بسته به نیازمندی محصول مفید است [18].
Cruise Control (https://github.com/linkedin/cruise-control) توسط developerهای LinkedIn ساخته شده. چون هزاران broker در clusterهایشان دارند، تجربهٔ اجرای cluster Kafka را دارند و برخی pain pointهای Kafka را سالها codify و automate کردهاند. REST API و UI هر دو برای تعامل با این ابزار در دسترس است. جالبترین featureها برای ما این است که Cruise Control cluster را watch میکند و بر اساس workload پیشنهاد rebalancing میدهد [19].
Confluent Control Center (https://docs.confluent.io/current/control-center/index.html) ابزار web-based دیگری برای monitor و manage cluster است. نکته این است که در حال حاضر feature commercial است و برای production setup به enterprise license نیاز دارد. اگر subscription Confluent Platform دارید، ارزش بررسی دارد. این ابزار از dashboard استفاده میکند و میتواند message failure، network latency و connectorهای خارجی را شناسایی کند.
بهطور کلی Kafka گزینههای زیادی برای manage و monitor cluster به ما میدهد. سیستمهای توزیعشده سختاند و هرچه تجربه بیشتر شود، skill و practiceهای monitoring بهتر میشوند.
خلاصه
- علاوه بر shell scriptهای همراه Kafka، administration client هم وجود دارد که API access به taskهای مهم مانند ایجاد topic میدهد.
- ابزارهایی مانند kcat و Confluent REST Proxy API راههایی برای تعامل developerها با cluster فراهم میکنند.
- اگرچه Kafka در هسته از log برای دادهٔ client استفاده میکند، logهای مختلف مخصوص بهرهبرداری broker وجود دارند که باید نگه داریم. باید مدیریت این logها (و logهای ZooKeeper) را برای جزئیات troubleshooting در صورت نیاز در نظر بگیریم.
- درک advertised listenerها میتواند رفتاری را که در ابتدا برای اتصال client ناهماهنگ به نظر میرسد توضیح دهد.
- Kafka از JMX برای metric استفاده میکند. metricها را از clientها (producer و consumer) و هم از brokerها میتوانید ببینید.
- میتوانیم producer و consumer interceptor برای پیادهسازی concernهای crosscutting استفاده کنیم. یک مثال افزودن tracing ID برای monitor تحویل پیام است.
منابع
- «Class AdminClient.» Confluent documentation (n.d.). https://docs.confluent.io/5.3.1/clients/javadocs/index.html?org/apache/kafka/clients/admin/AdminClient.html (accessed November 17, 2020).
- «kcat.» GitHub. https://github.com/edenhill/kcat/#readme (accessed August 25, 2021).
- «Kafka Security & the Confluent Platform.» Confluent documentation (n.d.). https://docs.confluent.io/2.0.1/kafka/platform-security.html#kafka-security-the-confluent-platform (accessed August 25, 2021).
- «Confluent REST APIs: Overview: Features.» Confluent documentation (n.d.). https://docs.confluent.io/platform/current/kafka-rest/index.html#features (accessed February 20, 2019).
- «REST Proxy Quick Start.» Confluent documentation (n.d.). https://docs.confluent.io/platform/current/kafka-rest/quickstart.html (accessed February 22, 2019).
- «Using Confluent Platform systemd Service Unit Files.» Confluent documentation (n.d.). https://docs.confluent.io/platform/current/installation/scripted-install.html#overview (accessed January 15, 2021).
- «Class RollingFileAppender.» Apache Software Foundation (n.d.). https://logging.apache.org/log4j/1.2/apidocs/org/apache/log4j/RollingFileAppender.html (accessed April 22, 2020).
- log4j.properties. Apache Kafka GitHub (March 26, 2020). https://github.com/apache/kafka/blob/99b9b3e84f4e98c3f07714e1de6a139a004cbc5b/config/log4j.properties (accessed June 17, 2020).
- «Running ZooKeeper in Production.» Confluent documentation (n.d.). https://docs.confluent.io/platform/current/zookeeper/deployment.html#running-zk-in-production (accessed July 23, 2021).
- «ZooKeeper Administrator's Guide.» Apache Software Foundation (n.d.). https://zookeeper.apache.org/doc/r3.4.5/zookeeperAdmin.html (accessed June 10, 2020).
- «ZooKeeper Getting Started Guide.» Apache Software Foundation (n.d.). https://zookeeper.apache.org/doc/r3.1.2/zookeeperStarted.html (accessed August 19, 2020).
- R. Moffatt. «Kafka Listeners – Explained.» Confluent blog (July 1, 2019). https://www.confluent.io/blog/kafka-listeners-explained/ (accessed June 11, 2020).
- «Kafka Monitoring and Metrics Using JMX.» Confluent documentation (n.d.). https://docs.confluent.io/platform/current/installation/docker/operations/monitoring.html (accessed June 12, 2020).
- «Monitoring Kafka: Broker Metrics.» Confluent documentation (n.d.). https://docs.confluent.io/5.4.0/kafka/monitoring.html#broker-metrics (accessed May 1, 2020).
- «Interface ProducerInterceptor.» Apache Software Foundation (n.d.). https://kafka.apache.org/27/javadoc/org/apache/kafka/clients/producer/ProducerInterceptor.html (accessed June 1, 2020).
- «Interface ConsumerInterceptor.» Apache Software Foundation (n.d.). https://kafka.apache.org/27/javadoc/org/apache/kafka/clients/consumer/ConsumerInterceptor.html (accessed June 1, 2020).
- «Monitoring.» Apache Software Foundation (n.d.). https://kafka.apache.org/documentation/#monitoring (accessed May 1, 2020).
- Yahoo CMAK README.md. GitHub (March 5, 2020). https://github.com/yahoo/CMAK/blob/master/README.md (accessed July 20, 2021).
- README.md. LinkedIn Cruise Control for Apache Kafka GitHub (June 30, 2021). https://github.com/linkedin/cruise-control/blob/migrate_to_kafka_2_4/README.md (accessed July 21, 2021).
بخش ۳ — فراتر رفتن
بخش ۳ بر چگونگی پیشبرد استفاده از Kafka فراتر از آنچه بخش ۲ بهعنوان قطعات هسته پوشش داد تمرکز دارد. در این بخش فراتر از داشتن cluster Kafka که بتوانیم داده بخوانیم و بنویسیم میرویم. امنیت بیشتر، schema داده و محصولات دیگر Kafka را اضافه میکنیم.
- در فصل ۱۰، cluster Kafka را با SSL، ACL و گزینههایی مانند quota تقویت میکنیم.
- در فصل ۱۱، Schema Registry و نحوهٔ کمک آن به evolution سازگار داده را بررسی میکنیم.
- در فصل ۱۲، Kafka Streams و ksqlDB را میبینیم.
این قطعات همگی بخشی از اکوسیستم Kafka هستند و لایههای abstraction بالاتری روی مباحث هستهای هستند که در بخش ۲ مطالعه کردید. در پایان این بخش آماده خواهید بود topicهای پیشرفتهتر Kafka را خودتان کاوش کنید و — بهتر از آن — بتوانید Kafka را در workflow روزمرهتان به کار ببرید.