Enable support for MQTT Parser in stirling #1756

ChinmayaSharma-hue · 2023-10-31T06:16:59Z

Summary: This PR adds the parser component of MQTT (v5), a newly added protocol.

Related issues: #341

Type of change: /kind feature

Test Plan: Added tests

ddelnano

@ChinmayaSharma-hue really appreciate all your hard work on this and very excited to have MQTT support within Pixie! I still haven't made it the whole way through this since I'm new to MQTT. However, I wanted to post the feedback I have so far.

ddelnano · 2023-10-31T16:36:31Z

src/stirling/source_connectors/socket_tracer/bcc_bpf/BUILD.bazel

@@ -82,6 +82,7 @@ pl_cc_test(
        "ENABLE_NATS_TRACING=true",
        "ENABLE_MONGO_TRACING=true",
        "ENABLE_AMQP_TRACING=true",
+        "ENABLE_MQTT_TRACING=true",
    ],
    deps = [
        "//src/stirling/bpf_tools/bcc_bpf:headers",


We should separate out any changes that aren't within the parsing code. This will be easier to review if we only have the following in this PR: protocols/mqtt/{types.h,parse.h,parser.cc} and any related build changes.

Sure, Noted! Will update in the next commit.

ddelnano · 2023-10-31T18:07:24Z

src/stirling/source_connectors/socket_tracer/protocols/mqtt/BUILD.bazel

+        "//src/common/json:cc_library",
+        "//src/common/zlib:cc_library",


Are these dependencies needed? I don't see any reference to json parsing or zlib

No, they are not, I just forgot to remove them when I copied the BUILD.bazel file from a different protocol. Will update in the next commit.

ddelnano · 2023-10-31T18:13:11Z

src/stirling/source_connectors/socket_tracer/protocols/mqtt/types.h

+struct State {
+  bool conn_closed = false;
+};
+
+struct StateWrapper {
+  State global;
+  std::monostate send;
+  std::monostate recv;
+};


Can you share more details on how this state structure will be used? Below NoState is indicated, so I wasn't sure if you see value in using state or if this was accidental.

I did this in the beginning when I wasn't sure if this was needed, will remove this.

ddelnano · 2023-10-31T18:18:05Z

src/stirling/source_connectors/socket_tracer/socket_trace_connector.cc

@@ -112,6 +112,9 @@ DEFINE_int32(stirling_enable_mux_tracing,
 DEFINE_int32(stirling_enable_amqp_tracing,
             gflags::Int32FromEnv("PX_STIRLING_ENABLE_AMQP_TRACING", px::stirling::TraceMode::On),
             "If true, stirling will trace and process AMQP messages.");
+DEFINE_int32(stirling_enable_mqtt_tracing,
+             gflags::Int32FromEnv("PX_STIRLING_ENABLE_MQTT_TRACING", px::stirling::TraceMode::On),


We should split this out to a later PR, but this should be px::stirling::TraceMode::OnForNewerKernels. Our 4.14 kernel build is close to the max BPF program instruction count and so new protocols can't be enabled wholesale.

ddelnano · 2023-10-31T18:26:14Z

src/stirling/source_connectors/socket_tracer/protocols/mqtt/parse.cc

+ParseState ParseFrame(message_type_t type, std::string_view* buf,
+                  Message* result) {
+    CTX_DCHECK(type == message_type_t::kRequest || type == message_type_t::kResponse);
+    if (buf->size() < 2) {


I assume this is 2 because we could have a 1 byte control header and 1 byte packet length? It might be easier to make this < 5 so that our remaining code can assume that the entire packet length will be accessible. Otherwise, we need to handle ExtractUVarInt errors differently depending on the context:

Buffer size: 3 bytes, Expected Packet length: 4 bytes -- binary_decoder->ExtraceUVarInt will fail and we need to return kNeedsMoreData

Buffer size: 5 bytes, Packet length payload: bogus UVarInt encoding -- binary_decoder->ExtraceUVarInt will fail and we need to return kInvalid

If we change the logic as I described above, any binary_decoder->ExtraceUVarInt error means that the UVarInt was larger than 4 bytes and is malformed or not MQTT data, so it can always be treated as kInvalid. We still need to check that the decoded value is within kMaxVarint28 that we discussed before, but it should simplify discerning the cases mentioned above.

Would it not work to return kNeedMoreData for all cases of ExtractUVarInt errors? If buffer size is 3 bytes and expected packet length is 4 then ExtractUVarInt would return an error and based on this kNeedsMoreData would be returned. I have added another function that checks whether or not the number returned by ExtractUVarInt is over 4 bytes, which would cause the function to return kInvalid. So both the cases would be taken care of in this way right?
Also, I am not sure why it would be easier to change <2 to <5. Wouldn't this eliminate cases where the buffer size is 2? (PINGREQ and PINRESP) are only 2 bytes, with remaining length set to 0.)

Are you talking about the case when ExtractUVarInt would return insufficient number of bytes error which would happen as it goes over 4 bytes to parse (as it does not know that the limit is 4 bytes), which would prompt my code to return kNeedsMoreData when in fact it is kInvalid?

Would it not work to return kNeedMoreData for all cases of ExtractUVarInt errors?

I don't think it will work because if only 2 bytes are available a UVarInt decode could return a complete value or it could return an insufficient number of bytes error. Since UVarInt's of 2-4 bytes in length are valid in MQTT, we can't use that error as an indicator unless we know that a 4 byte UVarInt is guaranteed to parse.

Also, I am not sure why it would be easier to change <2 to <5. Wouldn't this eliminate cases where the buffer size is 2? (PINGREQ and PINRESP) are only 2 bytes, with remaining length set to 0.)

Good call. Since PINGREQ and PINGRESP are a maximum of 2 bytes, that would be a problem and is another case we need to handle.

Are you talking about the case when ExtractUVarInt would return insufficient number of bytes error which would happen as it goes over 4 bytes to parse

Correct, that would be similar to case 1 from my original comment. If we check for 5 bytes, that would allow us to treat any UVarInt decoding as kInvalid since it would guarantee that it would be 5+ bytes in size.

Could we handle the PINGREQ and PINGRESP cases once the control_packet_code_flags variable is populated? Then we can check to see if the buffer contains 5 bytes? I'm thinking something like the following:

if (buf->size() < 2) { return ParseState::kNeedsMoreData; } PX_ASSIGN_OR_RETURN_ERROR(uint8_t control_packet_code_flags, decoder.ExtractBEInt<uint8_t>()); uint8_t control_packet_code = control_packet_code_flags >> 4; uint8_t control_packet_flags = control_packet_code_flags & 0x0F; MqttControlPacketType control_packet_type = GetControlPacketType(control_packet_code); result->control_packet_type = ControlPacketTypeStrings[control_packet_type]; if (control_packet_type == MqttControlPacketType::PINGREQ || control_packet_type == MqttControlPacketType::PINGRESP) { // Decode UVarInt, check it's 1 byte in length and return success // if decoding fails or it is > 1 byte, return kInvalid } // With the control messages less than 5 bytes in size handled, we can do the following validation. // This would then allow us to treat any insufficient byte errors from UVarInt decoding as kInvalid since // there shouldn't be a UVarInt with more than 4 bytes returned. if (buf->size() < 5) { return ParseState::kNeedsMoreData; } PX_ASSIGN_OR(size_t remaining_length, decoder.ExtractUVarInt(), return ParseState::kInvalid)

Ok, now I seem to have understood your point. Just to confirm, I have drawn a diagram to find out if we are on the same page.

You're talking about the circled out cases in the figure where ExtractUVarInt returns insufficient number of bytes error even when the buffer is complete (meaning that the full packet is present) due to incorrect encoding leading ExtractUVarInt to think there is more variable encoded data than there is. (I guess this would cause a repeated return of kNeedsMoreData to keep filling the buffer with the wrong packet data as it is invalid)
Also buf->size() < 5 returning kNeedsMoreData would work if UVarInt is more than one byte because remaining length being 2 byte or more would just mean that the buffer would definitely be bigger than 5.
So what you would be doing is just eliminate all cases of kNeedsMoreData before parsing UVarInt so that all the remaining cases of insufficient number of bytes error could be kInvalid.

This leaves all the cases where remaining length is one byte and value of remaining length is less than 4. Like for PUBACK remaining length is 3, and the remaining length itself takes one byte, so it is less than 5 bytes which would cause the code to return kNeedsMoreData.
One way to resolve this is after checking if buffer size is less than 5, we can extract the UVarInt remaining length and check if it is one byte and then proceed.

If ExtractUVarInt returns an integer more than one byte (but less than 4 bytes), that can be one of two things,

Encoding is wrong, kInvalid needs to be returned.

Buffer is incomplete, kNeedsMoreData needs to be returned.

If ExtractUVarInt returns insufficient number of bytes error then it is definitely kInvalid as it is trying to consume more than 4 bytes for UVarInt.

So in the first point there needs to be a differentiation for the two cases. At this point I am at a loss as to how to do this. If the remaining length bytes itself are wrong (encoding is wrong) then there is no way to validate whether or not the buffer is incomplete or the remaining length bytes are wrong.

So currently I have decided to do this,

// Decoding the variable encoding of remaining length field size_t remaining_length; if (control_packet_type == MqttControlPacketType::PINGREQ || control_packet_type == MqttControlPacketType::PINGRESP) { PX_ASSIGN_OR_RETURN_INVALID(remaining_length, decoder.ExtractUVarInt()); if (remaining_length > 0) { return ParseState::kInvalid; } } // Eliminating cases where kNeedsMoreData needs to be returned // If buffer size is less tan 4, there are chances that the remaining length is not present in its entirety if (decoder.BufSize() < 4) { // Checking if buffer is complete PX_ASSIGN_OR_RETURN_NEEDS_MORE_DATA(remaining_length, decoder.ExtractUVarInt()); // if remaining length is greater than 3 (4 if remaining length is included), then incomplete buffer, otherwise buffer is complete if (remaining_length > 3) { return ParseState::kNeedsMoreData; } } else { PX_ASSIGN_OR_RETURN_INVALID(remaining_length, decoder.ExtractUVarInt()); if (!VariableEncodingNumBytes(remaining_length).ok()) { return ParseState::kInvalid; } } // Making sure buffer is complete according to remaining length if (decoder.BufSize() < remaining_length) { return ParseState::kNeedsMoreData; }

This is not perfect since it can cause problems in this section,

// if remaining length is greater than 3 (4 if remaining length is included), then incomplete buffer, otherwise buffer is complete if (remaining_length > 3) { return ParseState::kNeedsMoreData; }

where I am deciding that buffer is incomplete if remaining length is greater than 3, when it could be that remaining length is greater than 3 simply because of an encoding error and the full data is still present in the buffer. But this works for most of the cases discussed.

Also <5 is replaced with <4 because the first byte is already extracted and I am using decoder's buffer.

I see. Thanks for the detailed explanation on those cases.

If we can special case any control codes that are known to be short in length (like we did with PINGREQ and PINGRESP, it may reduce the cases of treating some of these incorrectly. For now I'm fine with keeping the implementation as is unless you see an opportunity for special casing any of them.

Special casing only works for PINGREQ and PINGRESP as we can guarantee their sizes to be below 5. With all the other control packets there are chances that properties are present in variable header that can make them bigger.

ddelnano · 2023-11-01T14:58:11Z

src/stirling/source_connectors/socket_tracer/protocols/mqtt/parse.cc

+            if (result->header_fields["password_flag"]) {
+                PX_ASSIGN_OR_RETURN_ERROR(size_t password_length, decoder->ExtractBEInt<uint16_t>());
+                PX_ASSIGN_OR_RETURN_ERROR(std::string_view password, decoder->ExtractString(password_length));
+                result->payload["password"] = std::string(password);


Do you think this field is important to capture? Since we are capable of capturing everything within a connection, I think it's best to avoid sensitive information when it's easily detected.

I think we should either replace it with a string of the same length "XXXXX" or just skip over the field after advancing the buffer.

ddelnano · 2023-11-01T15:18:38Z

src/stirling/source_connectors/socket_tracer/protocols/mqtt/parse.cc

+        result->header_fields["dup"] = (control_packet_flags >> 3) != 0;
+        result->header_fields["retain"] = (control_packet_flags & 0x1) != 0;
+        result->header_fields["qos"] = (control_packet_flags >> 1) & 0x3;


Since these are expected to be on every message, I think these would be better as individual bools on the Message struct.

Except for QOS right? Because it would be helpful to know the exact QOS (0,1 or 2) instead of just knowing whether or not the qos is 0 or not.

Yea, I glossed over that the qos field wasn't a bool. The data type should fit whatever we need to store for each field.

ddelnano · 2023-11-01T16:03:32Z

src/stirling/source_connectors/socket_tracer/protocols/mqtt/stitcher.h

+inline RecordsWithErrorCount<mqtt::Record> StitchFrames(std::deque<mqtt::Message>* req_messages,
+                                                        std::deque<mqtt::Message>* resp_messages,
+                                                        NoState* /*state*/) {
+  return StitchMessagesWithTimestampOrder<mqtt::Record>(req_messages, resp_messages);


The stitcher should be the next PR after the parser changes, so I don't want to dive into this too much now. However, it seems that message ordering is only guaranteed within a given QoS level. I think using StitchMessagesWithTimestampOrder will result in invalid protocol traces if a client used multiple QoS levels.

In the next PR we will add tests for the Stitcher and we can model that situation. We will likely need to perform a similar process as StitchMessagesWithTimestampOrder, making sure that we only match frames within the same QoS since that will guarantee its assumptions hold. This is why it doesn't work with HTTP pipelining.

We can either use the QoS field as the map key to our new stitcher interface. This hasn't been documented yet as it's in the process of being merged (#1716) or we can leverage the protocol state to make sure we have the ordering correct. My initial thinking is that the former would be best, but we will have to see.

Makes sense. I had initially used StitchMessagesWithTimestampOrder as a placeholder as I was not sure what would work. I will focus more on the stitcher in the stitcher PR. Thanks for the additional context, it's very helpful.

ddelnano · 2023-11-01T16:35:02Z

src/stirling/source_connectors/socket_tracer/protocols/mqtt/parse_test.cc

+    ParseState result_state;
+    std::string_view frame_view;
+
+    uint8_t payload_format_indicator_publish[] = {


Did these payloads come from real packet captures or are they handcrafted?

They came from real packet captures. I used mosquitto to generate packets with different properties and payloads.

ddelnano · 2023-11-01T16:40:05Z

src/stirling/source_connectors/socket_tracer/protocols/mqtt/parse_test.cc

+    EXPECT_EQ(frame.header_fields["remaining_length"], (size_t) 16);
+    EXPECT_EQ(frame.header_fields["username_flag"], (unsigned long) 0);
+    EXPECT_EQ(frame.header_fields["password_flag"], (unsigned long) 0);
+    EXPECT_EQ(frame.header_fields["will_retain"], (unsigned long) 0);
+    EXPECT_EQ(frame.header_fields["will_qos"], (unsigned long) 0);
+    EXPECT_EQ(frame.header_fields["will_flag"], (unsigned long) 0);
+    EXPECT_EQ(frame.header_fields["clean_start"], (unsigned long) 1);
+    EXPECT_EQ(frame.header_fields["keep_alive"], (unsigned long) 60);


Please use UL to form the appropriate literals or remove the casting entirely. The tests seem to pass without these, so my preference is the latter. This applies to the other casting done in this file.

ddelnano · 2023-11-08T12:57:20Z

src/stirling/binaries/stirling_wrapper.cc

@@ -62,7 +62,7 @@ DEFINE_string(trace, "",
              "Dynamic trace to deploy. Either (1) the path to a file containing PxL or IR trace "
              "spec, or (2) <path to object file>:<symbol_name> for full-function tracing.");
 DEFINE_string(print_record_batches,
-              "http_events,mysql_events,pgsql_events,redis_events,cql_events,dns_events",
+              "http_events,mysql_events,pgsql_events,redis_events,cql_events,dns_events,mqtt_events",


We should hold off on updating this until the MQTT changes are released (one of the final changes).

ddelnano · 2023-11-08T12:57:27Z

src/stirling/source_connectors/socket_tracer/protocols/BUILD.bazel

@@ -45,5 +45,6 @@ pl_cc_library(
        "//src/stirling/source_connectors/socket_tracer/protocols/nats:cc_library",
        "//src/stirling/source_connectors/socket_tracer/protocols/pgsql:cc_library",
        "//src/stirling/source_connectors/socket_tracer/protocols/redis:cc_library",
+        "//src/stirling/source_connectors/socket_tracer/protocols/mqtt:cc_library",


This should be saved for a later change. Most likely once we introduce the "mqtt trace bpf test"

ddelnano · 2023-11-08T13:05:28Z

src/stirling/source_connectors/socket_tracer/socket_trace_connector.h

@@ -65,6 +65,7 @@ DECLARE_int32(stirling_enable_nats_tracing);
 DECLARE_int32(stirling_enable_kafka_tracing);
 DECLARE_int32(stirling_enable_mux_tracing);
 DECLARE_int32(stirling_enable_amqp_tracing);
+DECLARE_int32(stirling_enable_mqtt_tracing);


Changes to this file should also come in a later PR (when the MQTT trace bpf test is added).

ddelnano · 2023-11-08T13:05:57Z

src/stirling/source_connectors/socket_tracer/protocols/types.h

@@ -49,7 +50,8 @@ using FrameDequeVariant = std::variant<std::monostate,
                                       std::deque<redis::Message>,
                                       std::deque<kafka::Packet>,
                                       std::deque<nats::Message>,
-                                       std::deque<amqp::Frame>>;
+                                       std::deque<amqp::Frame>,
+                                       std::deque<mqtt::Message>>;


Changes to this file should also come in a later PR (when the MQTT trace bpf test is added).

ddelnano · 2023-11-08T13:06:26Z

src/stirling/source_connectors/socket_tracer/protocols/mqtt/types.h

+//-----------------------------------------------------------------------------
+
+/**
+ *  Record is the primary output of the http stitcher.


Suggested change

* Record is the primary output of the http stitcher.

* Record is the primary output of the MQTT stitcher.

ddelnano · 2023-11-08T13:07:03Z

src/stirling/source_connectors/socket_tracer/protocols/stitchers.h

@@ -30,3 +30,4 @@
 #include "src/stirling/source_connectors/socket_tracer/protocols/nats/stitcher.h"  // IWYU pragma: export
 #include "src/stirling/source_connectors/socket_tracer/protocols/pgsql/stitcher.h"  // IWYU pragma: export
 #include "src/stirling/source_connectors/socket_tracer/protocols/redis/stitcher.h"  // IWYU pragma: export
+#include "src/stirling/source_connectors/socket_tracer/protocols/mqtt/stitcher.h" // IWYU pragma: export


We should save this for the stitcher PR.

ddelnano · 2023-11-08T13:07:28Z

src/stirling/source_connectors/socket_tracer/protocols/mqtt/parse.h

+namespace protocols {
+
+/**
+ * Parses a single HTTP message from the input string.


Suggested change

* Parses a single HTTP message from the input string.

* Parses a single MQTT message from the input string.

ddelnano · 2023-11-08T13:21:47Z

src/stirling/source_connectors/socket_tracer/protocols/mqtt/parse.cc

+#include "src/stirling/utils/binary_decoder.h"
+#include "src/stirling/source_connectors/socket_tracer/protocols/mqtt/types.h"
+
+#define PX_ASSIGN_OR_RETURN_ERROR(expr, val_or) \


Can we rename this to PX_ASSIGN_OR_RETURN_NEEDS_MORE_DATA? Imo kNeedsMoreData isn't an error because it's an indicator that we should continue to retry. We also have the same macro defined in the pgsql parser and this would ensure our naming is consistent (source).

In addition to this, I believe we are returning kNeedsMoreData in places that we shouldn't. Ideally we would validate that the parser has the entire payload as early as possible and return kInvalid for any subsequent decoding. These later decodings shouldn't fail because we've validated that the buffer contains enough bytes, but imo it's the correct ParseState to return. The MQTT case is a little complex because we have multiple "payload" lengths -- remaining_length and variable_header_length.

My understanding is that remaining length will include the size of the entire MQTT frame. Assuming that's correct, we should only consider returning kNeedsMoreData until we can validate that the buffer is greater than or equal to remaining length. After that point, any decoding errors should be kInvalid.

It appears the variable length header can contain optional fields. We only decode these fields when we know that they should be present in order to maintain the assumption I mentioned in the previous paragraph. It seems we are already accomplishing that (as seen in the PUBCOMP case), so that shouldn't require any changes.

So what that would mean is only in the main parsing function before parsing the variable header or the payload, checks are done to make sure remaining length is equal to the buffer size and then for any subsequent bit extractions, kInvalid should be returned?
Right now for every extraction error kNeedsMoreData is being returned. I realise that extraction error happening is unlikely (unless remaining length field is incorrect) because we have already validated that the remaining length is equal to the buffer size so the full packet is already in the buffer. Would you suggest having PX_ASSIGN_OR_RETURN_INVALID to return kInvalid in variable header parsing and payload parsing? Because an extraction error in variable header/payload parsing, after we have already validated remaining length to be equal to buffer size, means that the remaining length field was wrong, which would make it an invalid MQTT packet.

Your latest changes are consistent with what I described above, so this lgtm.

ddelnano · 2023-11-08T14:00:48Z

src/stirling/source_connectors/socket_tracer/protocols/mqtt/types.h

+struct Message: public FrameBase {
+    message_type_t type = message_type_t::kUnknown;
+
+    std::string control_packet_type = "UNKNOWN";


We should try to avoid storing strings in the data table unless it's necessary. Since this field can be modeled as an 8 bit int we should use that instead (MqttControlPacketType / uint8_t).

We typically add a pxl function to map this int back to a string for queries (docs). This allows us to minimize the storage needed for the data type while still allowing the human readable name to be used for visualizations.

ddelnano · 2023-11-08T14:08:06Z

src/stirling/source_connectors/socket_tracer/protocols/mqtt/types.h

+    template<typename KeyType, typename ValueType>
+    static std::string MapToString(const std::map<KeyType, ValueType>& inputMap) {
+        std::string result = "{";
+        for (const auto& entry : inputMap) {
+            result += entry.first + ": ";
+            if constexpr (std::is_same_v<ValueType, uint32_t>) {
+                result += std::to_string(entry.second);
+            } else if constexpr (std::is_same_v<ValueType, std::string>) {
+                result += entry.second;
+            }
+            result += ", ";
+        }
+        if (!inputMap.empty()) {
+            result = result.substr(0, result.size() - 2); // Remove the trailing ", "
+        }
+        result += "}";
+        return result;
+    }


Could this be replaced with the ToJSONString function? It seems like this is similar to json encoding and we should use rapidjson rather than building it by hand.

ddelnano · 2023-11-08T14:08:23Z

src/stirling/source_connectors/socket_tracer/protocols/mqtt/types.h

+        std::string header_fields_str = "{";
+        for (const auto& entry : properties) {
+            header_fields_str += entry.first + ": " + std::string(entry.second) + ", ";
+        }
+        header_fields_str += "}";
+
+        std::string properties_str = "{";
+        for (const auto& entry : properties) {
+            properties_str += entry.first + ": " + entry.second + ", ";
+        }
+        properties_str += "}";
+
+        std::string payload_str = "{";
+        for (const auto& entry : properties) {
+            payload_str += entry.first + ": " + entry.second + ", ";
+        }
+        payload_str += "}";


Same question regarding json encoding here.

ddelnano · 2023-11-08T14:33:20Z

@ChinmayaSharma-hue just posted my second round of feedback and this is shaping up very nicely! Thanks again for all the hard work on this!

In addition to the comments, can you please update the PR description to match our GitHub template? This would have been surfaced once the GitHub actions for this PR are permitted to run (I don't have that permission, but I can get that triggered today).

ChinmayaSharma-hue · 2023-11-14T21:03:47Z

@ChinmayaSharma-hue I believe this will be my last round of comments. I know there have been a variety of things to address, but making this initial implementation solid will be worth it in the long run!

Also please check the linter build errors. Those will need to be resolved as well.

I haven't gotten to making all the required changes as of yet including the linter error fixes, I will let you know as soon as all the changes are made. There were some instances where I made silly errors which should have been fixed without alerting on your part. I hope to make less of these in the future! Thanks a lot for the guidance and support.

ddelnano

One comment regarding one of the changes from the previous round of feedback and two suggestions for fixing the runtime/int linter warnings.

ddelnano · 2023-11-15T17:59:23Z

src/stirling/source_connectors/socket_tracer/protocols/mqtt/parse.cc

+        PX_ASSIGN_OR_RETURN_INVALID(uint16_t topic_alias, decoder->ExtractBEInt<uint8_t>());
+        result->properties["maximum_qos"] = std::to_string(topic_alias);


Suggested change

PX_ASSIGN_OR_RETURN_INVALID(uint16_t topic_alias, decoder->ExtractBEInt<uint8_t>());

result->properties["maximum_qos"] = std::to_string(topic_alias);

PX_ASSIGN_OR_RETURN_INVALID(uint8_t max_qos, decoder->ExtractBEInt<uint8_t>());

result->properties["maximum_qos"] = std::to_string(max_qos);

ddelnano · 2023-11-15T18:16:00Z

src/stirling/source_connectors/socket_tracer/protocols/mqtt/parse.cc

+constexpr int kMaxVarInt24 = 2097152;
+constexpr int kMaxVarInt32 = 268435456;
+
+static inline StatusOr<size_t> VariableEncodingNumBytes(unsigned long integer) {


Suggested change

static inline StatusOr<size_t> VariableEncodingNumBytes(unsigned long integer) {

static inline StatusOr<size_t> VariableEncodingNumBytes(uint64_t integer) {

ddelnano · 2023-11-15T18:18:05Z

src/stirling/source_connectors/socket_tracer/protocols/mqtt/parse.cc

+        break;
+      }
+      case PropertyCode::SubscriptionIdentifier: {
+        PX_ASSIGN_OR_RETURN_INVALID(unsigned long subscription_id, decoder->ExtractUVarInt());


Suggested change

PX_ASSIGN_OR_RETURN_INVALID(unsigned long subscription_id, decoder->ExtractUVarInt());

PX_ASSIGN_OR_RETURN_INVALID(uint64_t subscription_id, decoder->ExtractUVarInt());

ChinmayaSharma-hue · 2023-11-18T04:01:35Z

I have fixed the linting errors (I ran arc lint locally), and I have fixed the other issues with the code. Sorry for the delay.

ddelnano · 2023-11-20T13:03:26Z

@pixie-io/maintainers can we kick off the github actions for this PR? This is ready for its final review and I will be approving once the build passes.

ddelnano

Thanks so much for this contribution @ChinmayaSharma-hue and looking forward to working together on the upcoming ones!

JamesMBartlett · 2023-11-29T04:31:39Z

Thanks for the contribution @ChinmayaSharma-hue. We require all of our commits to be GPG signed. Could you please follow this guide and sign your commits: https://docs.github.com/en/authentication/managing-commit-signature-verification/signing-commits

Signed-off-by: Chinmay <chinmaysharma1020@gmail.com>

…e tests Signed-off-by: Chinmay <chinmaysharma1020@gmail.com>

Signed-off-by: Chinmay <chinmaysharma1020@gmail.com>

ChinmayaSharma-hue · 2023-11-29T05:18:48Z

The commits are now gpg signed and verified.

ChinmayaSharma-hue had a problem deploying to pr-actions-approval October 31, 2023 06:17 — with GitHub Actions Error

ChinmayaSharma-hue changed the title ~~Extend ExtractUVarInt to support customizable kMaxVarintLen64~~ Enable support for MQTT Parser in stirling Oct 31, 2023

ddelnano mentioned this pull request Oct 31, 2023

Extend ExtractUVarInt to support customizable kMaxVarintLen64 #1757

Closed

ddelnano reviewed Nov 1, 2023

View reviewed changes

ChinmayaSharma-hue temporarily deployed to pr-actions-approval November 7, 2023 18:27 — with GitHub Actions Inactive

ddelnano reviewed Nov 8, 2023

View reviewed changes

ChinmayaSharma-hue had a problem deploying to pr-actions-approval November 9, 2023 10:01 — with GitHub Actions Error

ChinmayaSharma-hue had a problem deploying to pr-actions-approval November 9, 2023 10:04 — with GitHub Actions Error

ChinmayaSharma-hue force-pushed the mqtt-tracing branch from ed808de to 3216ad8 Compare November 9, 2023 10:08

ChinmayaSharma-hue had a problem deploying to pr-actions-approval November 9, 2023 10:09 — with GitHub Actions Error

ChinmayaSharma-hue had a problem deploying to pr-actions-approval November 9, 2023 10:13 — with GitHub Actions Error

ChinmayaSharma-hue had a problem deploying to pr-actions-approval November 9, 2023 10:16 — with GitHub Actions Error

ChinmayaSharma-hue force-pushed the mqtt-tracing branch from d3583eb to 73132dd Compare November 9, 2023 11:48

ChinmayaSharma-hue had a problem deploying to pr-actions-approval November 9, 2023 11:48 — with GitHub Actions Error

ChinmayaSharma-hue closed this Nov 9, 2023

ChinmayaSharma-hue force-pushed the mqtt-tracing branch from 73132dd to 56033f3 Compare November 9, 2023 11:49

ChinmayaSharma-hue had a problem deploying to pr-actions-approval November 9, 2023 11:50 — with GitHub Actions Error

ChinmayaSharma-hue reopened this Nov 9, 2023

ChinmayaSharma-hue had a problem deploying to pr-actions-approval November 9, 2023 11:57 — with GitHub Actions Error

ChinmayaSharma-hue had a problem deploying to pr-actions-approval November 14, 2023 19:40 — with GitHub Actions Error

ChinmayaSharma-hue had a problem deploying to pr-actions-approval November 15, 2023 17:46 — with GitHub Actions Error

ddelnano reviewed Nov 15, 2023

View reviewed changes

ChinmayaSharma-hue temporarily deployed to pr-actions-approval November 18, 2023 04:01 — with GitHub Actions Inactive

ddelnano requested a review from a team November 21, 2023 18:37

ddelnano approved these changes Nov 21, 2023

View reviewed changes

ddelnano requested a review from a team November 21, 2023 18:37

JamesMBartlett approved these changes Nov 29, 2023

View reviewed changes

ChinmayaSharma-hue added 13 commits November 29, 2023 10:43

Added MQTT Parser and corresponding tests

68c747b

Signed-off-by: Chinmay <chinmaysharma1020@gmail.com>

Changes to parser to return kInvalid wherever appropriate and annotat…

3c53e85

…e tests Signed-off-by: Chinmay <chinmaysharma1020@gmail.com>

VariableEncodingNumBytes addition and non parser related files removal

797ba64

Signed-off-by: Chinmay <chinmaysharma1020@gmail.com>

Non Parser related changes removal and minor code fixes

d25f447

Signed-off-by: Chinmay <chinmaysharma1020@gmail.com>

Non Parser related changes removal

d51b8a8

Signed-off-by: Chinmay <chinmaysharma1020@gmail.com>

Clang-format code fix

2c0075a

Signed-off-by: Chinmay <chinmaysharma1020@gmail.com>

Updated MQTT types file

8c1d8f0

Signed-off-by: Chinmay <chinmaysharma1020@gmail.com>

Updated MQTT parse header file

3024b1f

Signed-off-by: Chinmay <chinmaysharma1020@gmail.com>

Fixed kNeedsMoreData and kInvalid in the right places

778e1e3

Signed-off-by: Chinmay <chinmaysharma1020@gmail.com>

Fixed kNeedsMoreData and kInvalid in the right places

c739a67

Signed-off-by: Chinmay <chinmaysharma1020@gmail.com>

Added AUTH and its corresponding tests

0bd1375

Signed-off-by: Chinmay <chinmaysharma1020@gmail.com>

Minor typo fixes and modifications

bf35978

Signed-off-by: Chinmay <chinmaysharma1020@gmail.com>

Fixed lint errors

5fc5413

Signed-off-by: Chinmay <chinmaysharma1020@gmail.com>

ChinmayaSharma-hue force-pushed the mqtt-tracing branch from 1bd3d50 to 5fc5413 Compare November 29, 2023 05:14

ChinmayaSharma-hue temporarily deployed to pr-actions-approval November 29, 2023 05:14 — with GitHub Actions Inactive

ChinmayaSharma-hue requested a review from JamesMBartlett December 1, 2023 18:11

JamesMBartlett merged commit b1aa1a6 into pixie-io:main Dec 5, 2023
29 checks passed

		"//src/common/json:cc_library",
		"//src/common/zlib:cc_library",

	* Record is the primary output of the http stitcher.
	* Record is the primary output of the MQTT stitcher.

	* Parses a single HTTP message from the input string.
	* Parses a single MQTT message from the input string.

		PX_ASSIGN_OR_RETURN_INVALID(uint16_t topic_alias, decoder->ExtractBEInt<uint8_t>());
		result->properties["maximum_qos"] = std::to_string(topic_alias);

	static inline StatusOr<size_t> VariableEncodingNumBytes(unsigned long integer) {
	static inline StatusOr<size_t> VariableEncodingNumBytes(uint64_t integer) {

	PX_ASSIGN_OR_RETURN_INVALID(unsigned long subscription_id, decoder->ExtractUVarInt());
	PX_ASSIGN_OR_RETURN_INVALID(uint64_t subscription_id, decoder->ExtractUVarInt());

Enable support for MQTT Parser in stirling #1756

Enable support for MQTT Parser in stirling #1756

Conversation

ChinmayaSharma-hue commented Oct 31, 2023 • edited Loading

ddelnano left a comment • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ChinmayaSharma-hue Nov 6, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ChinmayaSharma-hue Nov 9, 2023 • edited Loading

Choose a reason for hiding this comment

ChinmayaSharma-hue Nov 9, 2023 • edited Loading

Choose a reason for hiding this comment

ChinmayaSharma-hue Nov 10, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ddelnano Nov 1, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ddelnano Nov 8, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ddelnano commented Nov 8, 2023

ChinmayaSharma-hue commented Nov 14, 2023

ddelnano left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ChinmayaSharma-hue commented Nov 18, 2023

ddelnano commented Nov 20, 2023

ddelnano left a comment

Choose a reason for hiding this comment

JamesMBartlett commented Nov 29, 2023

ChinmayaSharma-hue commented Nov 29, 2023

ChinmayaSharma-hue commented Oct 31, 2023 •

edited

Loading

ddelnano left a comment •

edited

Loading

ChinmayaSharma-hue Nov 6, 2023 •

edited

Loading

ChinmayaSharma-hue Nov 9, 2023 •

edited

Loading

ChinmayaSharma-hue Nov 9, 2023 •

edited

Loading

ChinmayaSharma-hue Nov 10, 2023 •

edited

Loading

ddelnano Nov 1, 2023 •

edited

Loading

ddelnano Nov 8, 2023 •

edited

Loading