Mux protocol support: parser

Summary:
Add support for Mux protocol.

From GH: pixie-io#327

Closes pixie-io#327

Test Plan: Frame parsing tests included.

Reviewers: #stirling, rcheng

Reviewed By: #stirling, rcheng

Subscribers: rcheng, yzhao

Signed-off-by: Dom Del Nano <ddelnano@gmail.com>

Differential Revision: https://phab.corp.pixielabs.ai/D10030

GitOrigin-RevId: f0206a1

.gitignore

@@ -36,6 +36,9 @@ bazel-*
 **/.vscode/.vs_code_bazel_build/**
 **/.vscode/ipch/
 
+# ctags file
+tags
+
 # Python cache files.
 .cache
 __pycache__
@@ -68,3 +71,7 @@ clang_tidy.log
 # LSIF data for code intel
 *.dump.lsif
 dump.lsif
+
+# Ignore files created by vagrant
+.vagrant/
+Vagrantfile

src/common/base/types.h

@@ -155,6 +155,18 @@ inline int operator<<(int24_t left, int shift) {
   return left.data;
 }
 
+struct __attribute__((packed)) uint24_t {
+  operator int() const { return data; }
+  uint24_t(int x) : data(x) {}
+  uint24_t() {}
+  uint32_t data : 24;
+};
+
+inline int operator<<(uint24_t left, int shift) {
+  left.data = (left.data << shift) & 0xffffff;
+  return left.data;
+}
+
 }  // namespace px
 
 // When used in a constexpr function, this will prevent compilation if assert does not pass.

src/common/base/types_test.cc

@@ -230,5 +230,20 @@ TEST(int24_t, VerifyInitializationAndBitShifting) {
   EXPECT_NE(val, 16777215);
 }
 
+TEST(uint24_t, VerifyInitializationAndBitShifting) {
+  EXPECT_EQ(sizeof(uint24_t), 3);
+
+  uint24_t t = 1;
+  EXPECT_EQ(t << 8, 256);
+
+  // Assign an int that uses each of the 3 bytes
+  uint24_t t2 = 0x10111;
+  EXPECT_EQ(t2 << 8, 0x11100);
+
+  uint24_t val = 0xffffff;
+  EXPECT_EQ(val, 0xffffff);
+  EXPECT_NE(val, -1);
+}
+
 }  // namespace const_types_test
 }  // namespace px

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

@@ -0,0 +1,52 @@
+# Copyright 2018- The Pixie Authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# SPDX-License-Identifier: Apache-2.0
+
+load("//bazel:pl_build_system.bzl", "pl_cc_library", "pl_cc_test")
+
+package(default_visibility = ["//src/stirling:__subpackages__"])
+
+pl_cc_library(
+    name = "cc_library",
+    srcs = glob(
+        [
+            "*.cc",
+        ],
+        exclude = [
+            "**/*_test.cc",
+        ],
+    ),
+    hdrs = glob(
+        [
+            "*.h",
+        ],
+    ),
+    deps = [
+        "//src/stirling/source_connectors/socket_tracer/protocols/common:cc_library",
+        "//src/stirling/utils:cc_library",
+    ],
+)
+
+pl_cc_test(
+    name = "parse_test",
+    srcs = ["parse_test.cc"],
+    deps = [":cc_library"],
+)
+
+pl_cc_test(
+    name = "types_test",
+    srcs = ["types_test.cc"],
+    deps = [":cc_library"],
+)

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

@@ -0,0 +1,135 @@
+/*
+ * Copyright 2018- The Pixie Authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#include <string>
+#include <utility>
+
+#include "src/stirling/source_connectors/socket_tracer/protocols/mux/parse.h"
+#include "src/stirling/utils/binary_decoder.h"
+
+namespace px {
+namespace stirling {
+namespace protocols {
+namespace mux {
+
+ParseState ParseFullFrame(BinaryDecoder* decoder, Frame* frame) {
+  PL_ASSIGN_OR(frame->tag, decoder->ExtractInt<uint24_t>(), return ParseState::kInvalid);
+
+  Type frame_type = static_cast<Type>(frame->type);
+
+  if (frame_type == Type::kRerrOld || frame_type == Type::kRerr) {
+    PL_ASSIGN_OR(std::string_view why, decoder->ExtractString(frame->MuxBodyLength()),
+                 return ParseState::kInvalid);
+    frame->why = std::string(why);
+    return ParseState::kSuccess;
+  }
+
+  if (frame_type == Type::kRinit || frame_type == Type::kTinit) {
+    // TODO(ddelnano): Add support for reading Tinit and Rinit compression, tls and other parameters
+    return ParseState::kSuccess;
+  }
+
+  if (frame_type == Type::kRdispatch) {
+    PL_ASSIGN_OR(frame->reply_status, decoder->ExtractInt<uint8_t>(), return ParseState::kInvalid);
+  }
+
+  PL_ASSIGN_OR(int16_t num_ctx, decoder->ExtractInt<int16_t>(), return ParseState::kInvalid);
+  absl::flat_hash_map<std::string, absl::flat_hash_map<std::string, std::string>> context;
+
+  // Parse the key, value context pairs supplied in the Rdispatch/Tdispatch messages.
+  // These entries pass distributed tracing context, request deadlines, client id context
+  // and retries among others.
+  for (int i = 0; i < num_ctx; i++) {
+    PL_ASSIGN_OR(size_t ctx_key_len, decoder->ExtractInt<int16_t>(), return ParseState::kInvalid);
+
+    PL_ASSIGN_OR(std::string_view ctx_key, decoder->ExtractString(ctx_key_len),
+                 return ParseState::kInvalid);
+
+    PL_ASSIGN_OR(size_t ctx_value_len, decoder->ExtractInt<int16_t>(), return ParseState::kInvalid);
+
+    absl::flat_hash_map<std::string, std::string> unpacked_value;
+    if (ctx_key == "com.twitter.finagle.Deadline") {
+      PL_ASSIGN_OR(int64_t timestamp, decoder->ExtractInt<int64_t>(), return ParseState::kInvalid);
+      PL_ASSIGN_OR(int64_t deadline, decoder->ExtractInt<int64_t>(), return ParseState::kInvalid);
+
+      unpacked_value["timestamp"] = std::to_string(timestamp / 1000);
+      unpacked_value["deadline"] = std::to_string(deadline / 1000);
+
+    } else if (ctx_key == "com.twitter.finagle.tracing.TraceContext") {
+      PL_ASSIGN_OR(int64_t span_id, decoder->ExtractInt<int64_t>(), return ParseState::kInvalid);
+      PL_ASSIGN_OR(int64_t parent_id, decoder->ExtractInt<int64_t>(), return ParseState::kInvalid);
+      PL_ASSIGN_OR(int64_t trace_id, decoder->ExtractInt<int64_t>(), return ParseState::kInvalid);
+      PL_ASSIGN_OR(int64_t flags, decoder->ExtractInt<int64_t>(), return ParseState::kInvalid);
+
+      unpacked_value["span id"] = std::to_string(span_id);
+      unpacked_value["parent id"] = std::to_string(parent_id);
+      unpacked_value["trace id"] = std::to_string(trace_id);
+      unpacked_value["flags"] = std::to_string(flags);
+
+    } else if (ctx_key == "com.twitter.finagle.thrift.ClientIdContext") {
+      PL_ASSIGN_OR(std::string_view ctx_value, decoder->ExtractString(ctx_value_len),
+                   return ParseState::kInvalid);
+      unpacked_value["name"] = std::string(ctx_value);
+
+    } else {
+      PL_ASSIGN_OR(std::string_view ctx_value, decoder->ExtractString(ctx_value_len),
+                   return ParseState::kInvalid);
+      unpacked_value["length"] = std::to_string(ctx_value.length());
+    }
+
+    context.insert({std::string(ctx_key), unpacked_value});
+  }
+
+  frame->context = std::move(context);
+
+  // TODO(ddelnano): Add dest and dtab parsing here
+  return ParseState::kSuccess;
+}
+
+}  // namespace mux
+
+template <>
+ParseState ParseFrame(message_type_t, std::string_view* buf, mux::Frame* frame, NoState*) {
+  BinaryDecoder decoder(*buf);
+
+  PL_ASSIGN_OR(frame->length, decoder.ExtractInt<int32_t>(), return ParseState::kInvalid);
+  if (frame->length > buf->length()) {
+    return ParseState::kNeedsMoreData;
+  }
+
+  PL_ASSIGN_OR(frame->type, decoder.ExtractInt<int8_t>(), return ParseState::kInvalid);
+  if (!mux::IsMuxType(frame->type)) {
+    return ParseState::kInvalid;
+  }
+
+  ParseState parse_state = mux::ParseFullFrame(&decoder, frame);
+  if (parse_state == ParseState::kSuccess) {
+    buf->remove_prefix(frame->length);
+  }
+  return parse_state;
+}
+
+template <>
+size_t FindFrameBoundary<mux::Frame>(message_type_t, std::string_view, size_t, NoState*) {
+  // Not implemented.
+  return std::string::npos;
+}
+
+}  // namespace protocols
+}  // namespace stirling
+}  // namespace px

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

@@ -0,0 +1,44 @@
+/*
+ * Copyright 2018- The Pixie Authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include "src/common/base/base.h"
+#include "src/stirling/source_connectors/socket_tracer/protocols/common/interface.h"
+#include "src/stirling/source_connectors/socket_tracer/protocols/mux/types.h"
+#include "src/stirling/utils/binary_decoder.h"
+
+namespace px {
+namespace stirling {
+namespace protocols {
+namespace mux {
+
+ParseState ParseFullFrame(BinaryDecoder* decoder, Frame* frame);
+
+}
+
+template <>
+ParseState ParseFrame(message_type_t type, std::string_view* buf, mux::Frame* frame, NoState*);
+
+template <>
+size_t FindFrameBoundary<mux::Frame>(message_type_t type, std::string_view buf, size_t start_pos,
+                                     NoState*);
+
+}  // namespace protocols
+}  // namespace stirling
+}  // namespace px