当前位置：网站首页>Pipy mqtt agent (III) logging

Pipy mqtt agent (III) logging

2022-04-21 09:07:00 【InfoQ】

Pipy MQTT Agent series ：

Current limiting and index statistics

Cluster load balancing

Nowadays, observability has become an important attribute of a product , It contains three directions ： journal 、 Tracking and measurement . stay

The first in the series

in , We simply realized metrics The statistics of , This time, the log is implemented . As for tracking ,MQTT 5 Version added

User Property

Support for , The extension information can be carried in the message header . about MQTT The tracking

At present, there is no standard in the industry

. If the end side can generate tracking data , The agent can perform transparent transmission and record it in the log . So in this article, we focus on logging and measurement .

Log is one of the common means to record the operation of the system , Through the log, we can record the behavior of the application in the past period of time , Then master the operation state of the system , Sometimes it can also be used as the basis for troubleshooting . Yes IoT In the scenario of Internet of things , The end-to-end troubleshooting is greatly limited , And there are a lot of them .

Realization

Pipy Stream processing is a network , Input the stream to a filter consisting of

The Conduit

in .

Current MQTT The agent has implemented current limiting 、 Index statistics and load balancing , It corresponds to three except the main entry script

modular

：

throttle

、

metrics

、

balancer

. The pipes in the modules are connected to each other , As shown in the figure below .

The requested flow is limited 、 Index statistics 、 After processing, the load balancing pipeline is sent to the upstream , The response flow returned from the upstream will be processed by the pipeline , And finally send it to the client .

If you want to record requests and responses , You need to connect the log pipeline to the processing flow of request and response . So , We added a new module

logger

, Define three pipes in the new module

request

、

response

and

logging

（ Refers to multiple pipes in a script ）.

seeing the name of a thing one thinks of its function , The first two pipelines are connected to the request and response processes respectively , The last one is to output the recorded information . It can be output to the console 、 Or something like that ClickHouse In storage （ It can reduce the data collection components , Directly into storage ）. Here is to demonstrate the function , Output information directly to the console .

The pipe name is unique in the same module , It can not be unique in multiple modules .

null

logger

As a part of message processing , The details of the request and response are recorded , Different from several other pipelines

Will not affect the convection itself

. Here we use Pipy A very important filter in

fork

filter

fork

It is a special connection filter ： It sends a copy of the input stream to another pipeline , The original flow will continue to be processed according to the original process , No convection . This feature just meets our needs .

Demo

Environmental Science

Continue to use

Before

Of MQTT broker colony ：

192.168.1.11： agent Pipy proxy

192.168.1.12： client emqx-bench

192.168.1.13： node EMQ X Broker

192.168.1.14： node EMQ X Broker

test

be familiar with MQTT Should be right about the message

qos

Not strange , Different settings （0、1、2）broker Different treatment of .

$ emqtt_bench pub -h 192.168.1.11 -p 1884 -c 1 -L 1 -I 1 -t bench/1 -m 'hello world!' -q 0

We use 1 Client publishing 1 Messages are verified separately qos Log collection of .

qos=0 journal

journal ：

2022-04-20 14:01:35.823 [INF] {&quot;clientID&quot;:&quot;ubuntu-dev2_bench_pub_1_3823694902&quot;,&quot;broker&quot;:null,&quot;clintAddress&quot;:&quot;192.168.1.12&quot;,&quot;clintPort&quot;:44726,&quot;timestamp&quot;:1650463292823,&quot;type&quot;:&quot;CONNECT&quot;,&quot;qos&quot;:0,&quot;dup&quot;:false,&quot;retain&quot;:false,&quot;protocolLevel&quot;:5,&quot;keepAlive&quot;:300,&quot;properties&quot;:{&quot;sessionExpiryInterval&quot;:0},&quot;cleanStart&quot;:true,&quot;direction&quot;:&quot;request&quot;}
2022-04-20 14:01:35.823 [INF] {&quot;clientID&quot;:&quot;ubuntu-dev2_bench_pub_1_3823694902&quot;,&quot;broker&quot;:&quot;192.168.1.13:1883&quot;,&quot;clintAddress&quot;:&quot;192.168.1.12&quot;,&quot;clintPort&quot;:44726,&quot;timestamp&quot;:1650463292834,&quot;type&quot;:&quot;CONNACK&quot;,&quot;qos&quot;:0,&quot;dup&quot;:false,&quot;retain&quot;:false,&quot;sessionPresent&quot;:false,&quot;reasonCode&quot;:0,&quot;properties&quot;:{&quot;maximumPacketSize&quot;:4096,&quot;retainAvailable&quot;:1,&quot;sharedSubscriptionAvailable&quot;:1,&quot;subscriptionIdentifierAvailable&quot;:1,&quot;topicAliasMaximum&quot;:65535,&quot;wildcardSubscriptionAvailable&quot;:1},&quot;direction&quot;:&quot;response&quot;}
2022-04-20 14:01:35.823 [INF] {&quot;clientID&quot;:&quot;ubuntu-dev2_bench_pub_1_3823694902&quot;,&quot;broker&quot;:&quot;192.168.1.13:1883&quot;,&quot;clintAddress&quot;:&quot;192.168.1.12&quot;,&quot;clintPort&quot;:44726,&quot;timestamp&quot;:1650463292835,&quot;type&quot;:&quot;PUBLISH&quot;,&quot;qos&quot;:0,&quot;dup&quot;:false,&quot;retain&quot;:false,&quot;topicName&quot;:&quot;bench/1&quot;,&quot;direction&quot;:&quot;request&quot;}

qos=1 journal

2022-04-20 14:05:11.315 [INF] {&quot;clientID&quot;:&quot;ubuntu-dev2_bench_pub_1_3598972520&quot;,&quot;broker&quot;:null,&quot;clintAddress&quot;:&quot;192.168.1.12&quot;,&quot;clintPort&quot;:44732,&quot;timestamp&quot;:1650463508315,&quot;type&quot;:&quot;CONNECT&quot;,&quot;qos&quot;:0,&quot;dup&quot;:false,&quot;retain&quot;:false,&quot;protocolLevel&quot;:5,&quot;keepAlive&quot;:300,&quot;properties&quot;:{&quot;sessionExpiryInterval&quot;:0},&quot;cleanStart&quot;:true,&quot;direction&quot;:&quot;request&quot;}
2022-04-20 14:05:11.315 [INF] {&quot;clientID&quot;:&quot;ubuntu-dev2_bench_pub_1_3598972520&quot;,&quot;broker&quot;:&quot;192.168.1.14:1883&quot;,&quot;clintAddress&quot;:&quot;192.168.1.12&quot;,&quot;clintPort&quot;:44732,&quot;timestamp&quot;:1650463508317,&quot;type&quot;:&quot;CONNACK&quot;,&quot;qos&quot;:0,&quot;dup&quot;:false,&quot;retain&quot;:false,&quot;sessionPresent&quot;:false,&quot;reasonCode&quot;:0,&quot;properties&quot;:{&quot;maximumPacketSize&quot;:4096,&quot;retainAvailable&quot;:1,&quot;sharedSubscriptionAvailable&quot;:1,&quot;subscriptionIdentifierAvailable&quot;:1,&quot;topicAliasMaximum&quot;:65535,&quot;wildcardSubscriptionAvailable&quot;:1},&quot;direction&quot;:&quot;response&quot;}
2022-04-20 14:05:11.315 [INF] {&quot;clientID&quot;:&quot;ubuntu-dev2_bench_pub_1_3598972520&quot;,&quot;broker&quot;:&quot;192.168.1.14:1883&quot;,&quot;clintAddress&quot;:&quot;192.168.1.12&quot;,&quot;clintPort&quot;:44732,&quot;timestamp&quot;:1650463508318,&quot;type&quot;:&quot;PUBLISH&quot;,&quot;qos&quot;:1,&quot;dup&quot;:false,&quot;retain&quot;:false,&quot;topicName&quot;:&quot;bench/1&quot;,&quot;packetIdentifier&quot;:2,&quot;direction&quot;:&quot;request&quot;}
2022-04-20 14:05:11.315 [INF] {&quot;clientID&quot;:&quot;ubuntu-dev2_bench_pub_1_3598972520&quot;,&quot;broker&quot;:&quot;192.168.1.14:1883&quot;,&quot;clintAddress&quot;:&quot;192.168.1.12&quot;,&quot;clintPort&quot;:44732,&quot;timestamp&quot;:1650463508318,&quot;type&quot;:&quot;PUBACK&quot;,&quot;qos&quot;:0,&quot;dup&quot;:false,&quot;retain&quot;:false,&quot;packetIdentifier&quot;:2,&quot;reasonCode&quot;:16,&quot;direction&quot;:&quot;response&quot;}

qos=2 journal

about qos by 2 Scene , We need another client to subscribe to messages .

$ emqtt_bench sub -h 192.168.1.11 -p 1884 -t bench/1 -c 1

2022-04-20 14:07:12.034 [INF] {&quot;clientID&quot;:&quot;ubuntu-dev2_bench_sub_1_1982626972&quot;,&quot;broker&quot;:null,&quot;clintAddress&quot;:&quot;192.168.1.12&quot;,&quot;clintPort&quot;:44734,&quot;timestamp&quot;:1650463629034,&quot;type&quot;:&quot;CONNECT&quot;,&quot;qos&quot;:0,&quot;dup&quot;:false,&quot;retain&quot;:false,&quot;protocolLevel&quot;:5,&quot;keepAlive&quot;:300,&quot;properties&quot;:{&quot;sessionExpiryInterval&quot;:0},&quot;cleanStart&quot;:true,&quot;direction&quot;:&quot;request&quot;}
2022-04-20 14:07:12.034 [INF] {&quot;clientID&quot;:&quot;ubuntu-dev2_bench_sub_1_1982626972&quot;,&quot;broker&quot;:&quot;192.168.1.13:1883&quot;,&quot;clintAddress&quot;:&quot;192.168.1.12&quot;,&quot;clintPort&quot;:44734,&quot;timestamp&quot;:1650463629036,&quot;type&quot;:&quot;CONNACK&quot;,&quot;qos&quot;:0,&quot;dup&quot;:false,&quot;retain&quot;:false,&quot;sessionPresent&quot;:false,&quot;reasonCode&quot;:0,&quot;properties&quot;:{&quot;maximumPacketSize&quot;:4096,&quot;retainAvailable&quot;:1,&quot;sharedSubscriptionAvailable&quot;:1,&quot;subscriptionIdentifierAvailable&quot;:1,&quot;topicAliasMaximum&quot;:65535,&quot;wildcardSubscriptionAvailable&quot;:1},&quot;direction&quot;:&quot;response&quot;}
2022-04-20 14:07:12.034 [INF] {&quot;clientID&quot;:&quot;ubuntu-dev2_bench_sub_1_1982626972&quot;,&quot;broker&quot;:&quot;192.168.1.13:1883&quot;,&quot;clintAddress&quot;:&quot;192.168.1.12&quot;,&quot;clintPort&quot;:44734,&quot;timestamp&quot;:1650463629036,&quot;type&quot;:&quot;SUBSCRIBE&quot;,&quot;qos&quot;:1,&quot;dup&quot;:false,&quot;retain&quot;:false,&quot;packetIdentifier&quot;:2,&quot;topicFilters&quot;:[{&quot;filter&quot;:&quot;bench/1&quot;,&quot;qos&quot;:0}],&quot;direction&quot;:&quot;request&quot;}
2022-04-20 14:07:12.034 [INF] {&quot;clientID&quot;:&quot;ubuntu-dev2_bench_sub_1_1982626972&quot;,&quot;broker&quot;:&quot;192.168.1.13:1883&quot;,&quot;clintAddress&quot;:&quot;192.168.1.12&quot;,&quot;clintPort&quot;:44734,&quot;timestamp&quot;:1650463629037,&quot;type&quot;:&quot;SUBACK&quot;,&quot;qos&quot;:0,&quot;dup&quot;:false,&quot;retain&quot;:false,&quot;packetIdentifier&quot;:2,&quot;reasonCodes&quot;:[],&quot;direction&quot;:&quot;response&quot;}
2022-04-20 14:07:12.034 [INF] :{}

Release the news ：

2022-04-20 14:08:57.739 [INF] [object pipy::mqtt::MessageHead]:{&quot;type&quot;:&quot;CONNECT&quot;,&quot;qos&quot;:0,&quot;dup&quot;:false,&quot;retain&quot;:false,&quot;protocolLevel&quot;:5,&quot;keepAlive&quot;:300,&quot;properties&quot;:{&quot;sessionExpiryInterval&quot;:0},&quot;clientID&quot;:&quot;ubuntu-dev2_bench_pub_1_1960359114&quot;,&quot;cleanStart&quot;:true,&quot;direction&quot;:&quot;request&quot;}
2022-04-20 14:08:57.744 [INF] [object pipy::mqtt::MessageHead]:{&quot;type&quot;:&quot;PUBLISH&quot;,&quot;qos&quot;:2,&quot;dup&quot;:false,&quot;retain&quot;:false,&quot;topicName&quot;:&quot;bench/1&quot;,&quot;packetIdentifier&quot;:2,&quot;direction&quot;:&quot;request&quot;}
2022-04-20 14:08:57.745 [INF] [object pipy::mqtt::MessageHead]:{&quot;type&quot;:&quot;PUBREL&quot;,&quot;qos&quot;:1,&quot;dup&quot;:false,&quot;retain&quot;:false,&quot;packetIdentifier&quot;:2,&quot;reasonCode&quot;:0,&quot;direction&quot;:&quot;request&quot;}
2022-04-20 14:09:00.739 [INF] {&quot;clientID&quot;:&quot;ubuntu-dev2_bench_pub_1_1960359114&quot;,&quot;broker&quot;:null,&quot;clintAddress&quot;:&quot;192.168.1.12&quot;,&quot;clintPort&quot;:44736,&quot;timestamp&quot;:1650463737739,&quot;type&quot;:&quot;CONNECT&quot;,&quot;qos&quot;:0,&quot;dup&quot;:false,&quot;retain&quot;:false,&quot;protocolLevel&quot;:5,&quot;keepAlive&quot;:300,&quot;properties&quot;:{&quot;sessionExpiryInterval&quot;:0},&quot;cleanStart&quot;:true,&quot;direction&quot;:&quot;request&quot;}
2022-04-20 14:09:00.739 [INF] {&quot;clientID&quot;:&quot;ubuntu-dev2_bench_pub_1_1960359114&quot;,&quot;broker&quot;:&quot;192.168.1.14:1883&quot;,&quot;clintAddress&quot;:&quot;192.168.1.12&quot;,&quot;clintPort&quot;:44736,&quot;timestamp&quot;:1650463737743,&quot;type&quot;:&quot;CONNACK&quot;,&quot;qos&quot;:0,&quot;dup&quot;:false,&quot;retain&quot;:false,&quot;sessionPresent&quot;:false,&quot;reasonCode&quot;:0,&quot;properties&quot;:{&quot;maximumPacketSize&quot;:4096,&quot;retainAvailable&quot;:1,&quot;sharedSubscriptionAvailable&quot;:1,&quot;subscriptionIdentifierAvailable&quot;:1,&quot;topicAliasMaximum&quot;:65535,&quot;wildcardSubscriptionAvailable&quot;:1},&quot;direction&quot;:&quot;response&quot;}
2022-04-20 14:09:00.739 [INF] {&quot;clientID&quot;:&quot;ubuntu-dev2_bench_pub_1_1960359114&quot;,&quot;broker&quot;:&quot;192.168.1.14:1883&quot;,&quot;clintAddress&quot;:&quot;192.168.1.12&quot;,&quot;clintPort&quot;:44736,&quot;timestamp&quot;:1650463737744,&quot;type&quot;:&quot;PUBLISH&quot;,&quot;qos&quot;:2,&quot;dup&quot;:false,&quot;retain&quot;:false,&quot;topicName&quot;:&quot;bench/1&quot;,&quot;packetIdentifier&quot;:2,&quot;direction&quot;:&quot;request&quot;}
2022-04-20 14:09:00.739 [INF] {&quot;clientID&quot;:&quot;ubuntu-dev2_bench_pub_1_1960359114&quot;,&quot;broker&quot;:&quot;192.168.1.14:1883&quot;,&quot;clintAddress&quot;:&quot;192.168.1.12&quot;,&quot;clintPort&quot;:44736,&quot;timestamp&quot;:1650463737745,&quot;type&quot;:&quot;PUBREC&quot;,&quot;qos&quot;:0,&quot;dup&quot;:false,&quot;retain&quot;:false,&quot;packetIdentifier&quot;:2,&quot;reasonCode&quot;:0,&quot;direction&quot;:&quot;response&quot;}
2022-04-20 14:09:00.739 [INF] {&quot;clientID&quot;:&quot;ubuntu-dev2_bench_sub_1_1982626972&quot;,&quot;broker&quot;:&quot;192.168.1.13:1883&quot;,&quot;clintAddress&quot;:&quot;192.168.1.12&quot;,&quot;clintPort&quot;:44734,&quot;timestamp&quot;:1650463737745,&quot;type&quot;:&quot;PUBLISH&quot;,&quot;qos&quot;:0,&quot;dup&quot;:false,&quot;retain&quot;:false,&quot;topicName&quot;:&quot;bench/1&quot;,&quot;direction&quot;:&quot;response&quot;}
2022-04-20 14:09:00.739 [INF] {&quot;clientID&quot;:&quot;ubuntu-dev2_bench_pub_1_1960359114&quot;,&quot;broker&quot;:&quot;192.168.1.14:1883&quot;,&quot;clintAddress&quot;:&quot;192.168.1.12&quot;,&quot;clintPort&quot;:44736,&quot;timestamp&quot;:1650463737745,&quot;type&quot;:&quot;PUBREL&quot;,&quot;qos&quot;:1,&quot;dup&quot;:false,&quot;retain&quot;:false,&quot;packetIdentifier&quot;:2,&quot;reasonCode&quot;:0,&quot;direction&quot;:&quot;request&quot;}

summary

This time, we introduced... Through plug-in

logger

modular , The information of request and response is recorded . Yes Pipy Come on , The proxy function is realized by introducing a new pipeline , The new pipe is connected to other pipes through a connecting filter , Access to the processing flow of the stream . Maybe this process is more abstract , A separate article will describe Pipy Treatment of convection .

版权声明
本文为[InfoQ]所创，转载请带上原文链接，感谢
https://yzsam.com/2022/04/202204210901540018.html