ObservabilityLogging: Add interface for logging and config parsing implementation for GCP observability (#31571)

* ObservabilityLogging: Add interface for logging and config parsing implementation for GCP observability

* Trailing new lines

* Fix naked include

* clang-tidy

* Reviewer comments

* Reviewer comments

src/cpp/ext/filters/logging/BUILD

@@ -0,0 +1,47 @@
+# gRPC Bazel BUILD file.
+#
+# Copyright 2022 gRPC 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.
+
+load(
+    "//bazel:grpc_build_system.bzl",
+    "grpc_cc_library",
+)
+
+licenses(["reciprocal"])
+
+package(
+    default_visibility = ["//visibility:public"],
+    features = [
+        "layering_check",
+    ],
+)
+
+grpc_cc_library(
+    name = "logging_sink",
+    hdrs = [
+        "logging_sink.h",
+    ],
+    external_deps = [
+        "absl/strings",
+    ],
+    language = "c++",
+    visibility = [
+        "//src/cpp/ext/gcp:__subpackages__",
+        "//test:__subpackages__",
+    ],
+    deps = [
+        "//:gpr_platform",
+    ],
+)

src/cpp/ext/filters/logging/logging_sink.h

@@ -0,0 +1,61 @@
+//
+//
+// Copyright 2022 gRPC 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.
+//
+//
+
+#ifndef GRPC_INTERNAL_CPP_EXT_FILTERS_LOGGING_LOGGING_SINK_H
+#define GRPC_INTERNAL_CPP_EXT_FILTERS_LOGGING_LOGGING_SINK_H
+
+#include <grpc/support/port_platform.h>
+
+#include "absl/strings/string_view.h"
+
+namespace grpc {
+namespace internal {
+
+// Interface for a logging sink that will be used by the logging filter.
+class LoggingSink {
+ public:
+  class Config {
+   public:
+    Config(uint32_t max_metadata_bytes, uint32_t max_message_bytes)
+        : max_metadata_bytes_(max_metadata_bytes),
+          max_message_bytes_(max_message_bytes) {}
+    bool MetadataLoggingEnabled() { return max_metadata_bytes_ != 0; }
+    bool MessageLoggingEnabled() { return max_message_bytes_ != 0; }
+    bool ShouldLog() {
+      return MetadataLoggingEnabled() || MessageLoggingEnabled();
+    }
+
+    bool operator==(const Config& other) const {
+      return max_metadata_bytes_ == other.max_metadata_bytes_ &&
+             max_message_bytes_ == other.max_message_bytes_;
+    }
+
+   private:
+    uint32_t max_metadata_bytes_;
+    uint32_t max_message_bytes_;
+  };
+
+  virtual ~LoggingSink() = default;
+
+  virtual Config FindMatch(bool is_client, absl::string_view path) = 0;
+};
+
+}  // namespace internal
+}  // namespace grpc
+
+#endif  // GRPC_INTERNAL_CPP_EXT_FILTERS_LOGGING_LOGGING_SINK_H

src/cpp/ext/gcp/BUILD

@@ -89,3 +89,21 @@ grpc_cc_library(
         "//src/core:validation_errors",
     ],
 )
+
+grpc_cc_library(
+    name = "observability_logging_sink",
+    srcs = [
+        "observability_logging_sink.cc",
+    ],
+    hdrs = [
+        "observability_logging_sink.h",
+    ],
+    language = "c++",
+    tags = ["nofixdeps"],
+    visibility = ["//test:__subpackages__"],
+    deps = [
+        ":observability_config",
+        "//:gpr_platform",
+        "//src/cpp/ext/filters/logging:logging_sink",
+    ],
+)

src/cpp/ext/gcp/observability_logging_sink.cc

@@ -0,0 +1,82 @@
+//
+//
+// Copyright 2022 gRPC 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.
+//
+//
+
+#include <grpc/support/port_platform.h>
+
+#include "src/cpp/ext/gcp/observability_logging_sink.h"
+
+#include <stddef.h>
+
+#include <algorithm>
+
+namespace grpc {
+namespace internal {
+
+ObservabilityLoggingSink::ObservabilityLoggingSink(
+    GcpObservabilityConfig::CloudLogging logging_config) {
+  for (auto& client_rpc_event_config : logging_config.client_rpc_events) {
+    client_configs.emplace_back(client_rpc_event_config);
+  }
+  for (auto& server_rpc_event_config : logging_config.server_rpc_events) {
+    server_configs.emplace_back(server_rpc_event_config);
+  }
+}
+
+LoggingSink::Config ObservabilityLoggingSink::FindMatch(
+    bool is_client, absl::string_view path) {
+  size_t pos = path.find('/');
+  if (pos == absl::string_view::npos) {
+    // bad path - did not find '/'
+    return LoggingSink::Config(0, 0);
+  }
+  absl::string_view service =
+      path.substr(0, pos);  // service name is before the '/'
+  absl::string_view method =
+      path.substr(pos + 1);  // method name starts after the '/'
+  const auto& configs = is_client ? client_configs : server_configs;
+  for (const auto& config : configs) {
+    for (const auto& config_method : config.parsed_methods) {
+      if ((config_method.service == "*") ||
+          ((service == config_method.service) &&
+           ((config_method.method == "*") ||
+            (method == config_method.method)))) {
+        if (config.exclude) {
+          return LoggingSink::Config(0, 0);
+        }
+        return LoggingSink::Config(config.max_metadata_bytes,
+                                   config.max_message_bytes);
+      }
+    }
+  }
+  return LoggingSink::Config(0, 0);
+}
+
+ObservabilityLoggingSink::Configuration::Configuration(
+    const GcpObservabilityConfig::CloudLogging::RpcEventConfiguration&
+        rpc_event_config)
+    : exclude(rpc_event_config.exclude),
+      max_metadata_bytes(rpc_event_config.max_metadata_bytes),
+      max_message_bytes(rpc_event_config.max_message_bytes) {
+  for (auto& parsed_method : rpc_event_config.parsed_methods) {
+    parsed_methods.emplace_back(ParsedMethod{
+        std::string(parsed_method.service), std::string(parsed_method.method)});
+  }
+}
+
+}  // namespace internal
+}  // namespace grpc

src/cpp/ext/gcp/observability_logging_sink.h

@@ -0,0 +1,70 @@
+//
+//
+// Copyright 2022 gRPC 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.
+//
+//
+
+#ifndef GRPC_INTERNAL_CPP_EXT_GCP_OBSERVABILITY_GCP_OBSERVABILITY_LOGGING_SINK_H
+#define GRPC_INTERNAL_CPP_EXT_GCP_OBSERVABILITY_GCP_OBSERVABILITY_LOGGING_SINK_H
+
+#include <grpc/support/port_platform.h>
+
+#include <stdint.h>
+
+#include <string>
+#include <vector>
+
+#include "absl/strings/string_view.h"
+
+#include "src/cpp/ext/filters/logging/logging_sink.h"
+#include "src/cpp/ext/gcp/observability_config.h"
+
+namespace grpc {
+namespace internal {
+
+// Interface for a logging sink that will be used by the logging filter.
+class ObservabilityLoggingSink : public LoggingSink {
+ public:
+  explicit ObservabilityLoggingSink(
+      GcpObservabilityConfig::CloudLogging logging_config);
+
+  ~ObservabilityLoggingSink() override = default;
+
+  LoggingSink::Config FindMatch(bool is_client,
+                                absl::string_view path) override;
+
+ private:
+  struct Configuration {
+    explicit Configuration(
+        const GcpObservabilityConfig::CloudLogging::RpcEventConfiguration&
+            rpc_event_config);
+    struct ParsedMethod {
+      std::string service;
+      std::string method;
+    };
+    std::vector<ParsedMethod> parsed_methods;
+    bool exclude = false;
+    uint32_t max_metadata_bytes = 0;
+    uint32_t max_message_bytes = 0;
+  };
+
+  std::vector<Configuration> client_configs;
+  std::vector<Configuration> server_configs;
+};
+
+}  // namespace internal
+}  // namespace grpc
+
+#endif  // GRPC_INTERNAL_CPP_EXT_GCP_OBSERVABILITY_GCP_OBSERVABILITY_LOGGING_SINK_H

test/cpp/ext/gcp/BUILD

@@ -47,3 +47,18 @@ grpc_cc_test(
         "//test/cpp/util:test_util",
     ],
 )
+
+grpc_cc_test(
+    name = "observability_logging_sink_test",
+    srcs = [
+        "observability_logging_sink_test.cc",
+    ],
+    external_deps = [
+        "gtest",
+    ],
+    language = "C++",
+    deps = [
+        "//src/cpp/ext/gcp:observability_logging_sink",
+        "//test/cpp/util:test_util",
+    ],
+)

test/cpp/ext/gcp/observability_logging_sink_test.cc

@@ -0,0 +1,306 @@
+//
+// Copyright 2022 gRPC 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.
+//
+
+#include "src/cpp/ext/gcp/observability_logging_sink.h"
+
+#include "gmock/gmock.h"
+#include "gtest/gtest.h"
+
+#include "test/core/util/test_config.h"
+
+namespace grpc {
+namespace internal {
+
+namespace {
+
+TEST(GcpObservabilityLoggingSinkTest, LoggingConfigEmpty) {
+  const char* json_str = R"json({
+      "cloud_logging": {
+      }
+    })json";
+  auto json = grpc_core::Json::Parse(json_str);
+  ASSERT_TRUE(json.ok()) << json.status();
+  grpc_core::ValidationErrors errors;
+  auto config = grpc_core::LoadFromJson<GcpObservabilityConfig>(
+      *json, grpc_core::JsonArgs(), &errors);
+  ASSERT_TRUE(errors.ok()) << errors.status("unexpected errors");
+  ObservabilityLoggingSink sink(config.cloud_logging.value());
+  // client test
+  EXPECT_EQ(sink.FindMatch(true, "foo/bar"), LoggingSink::Config(0, 0));
+  // server test
+  EXPECT_EQ(sink.FindMatch(false, "foo/bar"), LoggingSink::Config(0, 0));
+}
+
+TEST(GcpObservabilityLoggingSinkTest, LoggingConfigClientWildCardEntries) {
+  const char* json_str = R"json({
+      "cloud_logging": {
+        "client_rpc_events": [
+          {
+            "methods": ["*"],
+            "max_metadata_bytes": 1024,
+            "max_message_bytes": 4096
+          }
+        ]
+      }
+    })json";
+  auto json = grpc_core::Json::Parse(json_str);
+  ASSERT_TRUE(json.ok()) << json.status();
+  grpc_core::ValidationErrors errors;
+  auto config = grpc_core::LoadFromJson<GcpObservabilityConfig>(
+      *json, grpc_core::JsonArgs(), &errors);
+  ASSERT_TRUE(errors.ok()) << errors.status("unexpected errors");
+  ObservabilityLoggingSink sink(config.cloud_logging.value());
+  // client test
+  EXPECT_EQ(sink.FindMatch(true, "foo/bar"), LoggingSink::Config(1024, 4096));
+  // server test
+  EXPECT_EQ(sink.FindMatch(false, "foo/bar"), LoggingSink::Config(0, 0));
+}
+
+TEST(GcpObservabilityLoggingSinkTest, LoggingConfigBadPath) {
+  const char* json_str = R"json({
+      "cloud_logging": {
+        "client_rpc_events": [
+          {
+            "methods": ["*"],
+            "max_metadata_bytes": 1024,
+            "max_message_bytes": 4096
+          }
+        ]
+      }
+    })json";
+  auto json = grpc_core::Json::Parse(json_str);
+  ASSERT_TRUE(json.ok()) << json.status();
+  grpc_core::ValidationErrors errors;
+  auto config = grpc_core::LoadFromJson<GcpObservabilityConfig>(
+      *json, grpc_core::JsonArgs(), &errors);
+  ASSERT_TRUE(errors.ok()) << errors.status("unexpected errors");
+  ObservabilityLoggingSink sink(config.cloud_logging.value());
+  EXPECT_EQ(sink.FindMatch(true, "foo"), LoggingSink::Config(0, 0));
+}
+
+TEST(GcpObservabilityLoggingSinkTest,
+     LoggingConfigClientWildCardServiceEntries) {
+  const char* json_str = R"json({
+      "cloud_logging": {
+        "client_rpc_events": [
+          {
+            "methods": ["service/*"],
+            "max_metadata_bytes": 1024,
+            "max_message_bytes": 4096
+          }
+        ]
+      }
+    })json";
+  auto json = grpc_core::Json::Parse(json_str);
+  ASSERT_TRUE(json.ok()) << json.status();
+  grpc_core::ValidationErrors errors;
+  auto config = grpc_core::LoadFromJson<GcpObservabilityConfig>(
+      *json, grpc_core::JsonArgs(), &errors);
+  ASSERT_TRUE(errors.ok()) << errors.status("unexpected errors");
+  ObservabilityLoggingSink sink(config.cloud_logging.value());
+  // client test
+  EXPECT_EQ(sink.FindMatch(true, "service/bar"),
+            LoggingSink::Config(1024, 4096));
+  EXPECT_EQ(sink.FindMatch(true, "foo/bar"), LoggingSink::Config(0, 0));
+  // server test
+  EXPECT_EQ(sink.FindMatch(false, "service/bar"), LoggingSink::Config(0, 0));
+  EXPECT_EQ(sink.FindMatch(false, "foo/bar"), LoggingSink::Config(0, 0));
+}
+
+TEST(GcpObservabilityLoggingSinkTest,
+     LoggingConfigClientMultipleMethodEntries) {
+  const char* json_str = R"json({
+      "cloud_logging": {
+        "client_rpc_events": [
+          {
+            "methods": ["foo/bar", "foo/baz"],
+            "max_metadata_bytes": 1024,
+            "max_message_bytes": 4096
+          }
+        ]
+      }
+    })json";
+  auto json = grpc_core::Json::Parse(json_str);
+  ASSERT_TRUE(json.ok()) << json.status();
+  grpc_core::ValidationErrors errors;
+  auto config = grpc_core::LoadFromJson<GcpObservabilityConfig>(
+      *json, grpc_core::JsonArgs(), &errors);
+  ASSERT_TRUE(errors.ok()) << errors.status("unexpected errors");
+  ObservabilityLoggingSink sink(config.cloud_logging.value());
+  // client test
+  EXPECT_EQ(sink.FindMatch(true, "foo/bar"), LoggingSink::Config(1024, 4096));
+  EXPECT_EQ(sink.FindMatch(true, "foo/baz"), LoggingSink::Config(1024, 4096));
+  // server test
+  EXPECT_EQ(sink.FindMatch(false, "foo/bar"), LoggingSink::Config(0, 0));
+  EXPECT_EQ(sink.FindMatch(false, "foo/baz"), LoggingSink::Config(0, 0));
+}
+
+TEST(GcpObservabilityLoggingSinkTest, LoggingConfigClientMultipleEventEntries) {
+  const char* json_str = R"json({
+      "cloud_logging": {
+        "client_rpc_events": [
+          {
+            "methods": ["foo/bar"],
+            "max_metadata_bytes": 1024,
+            "max_message_bytes": 4096
+          },
+          {
+            "methods": ["foo/baz"],
+            "max_metadata_bytes": 512,
+            "max_message_bytes": 2048
+          }
+        ]
+      }
+    })json";
+  auto json = grpc_core::Json::Parse(json_str);
+  ASSERT_TRUE(json.ok()) << json.status();
+  grpc_core::ValidationErrors errors;
+  auto config = grpc_core::LoadFromJson<GcpObservabilityConfig>(
+      *json, grpc_core::JsonArgs(), &errors);
+  ASSERT_TRUE(errors.ok()) << errors.status("unexpected errors");
+  ObservabilityLoggingSink sink(config.cloud_logging.value());
+  // client test
+  EXPECT_EQ(sink.FindMatch(true, "foo/bar"), LoggingSink::Config(1024, 4096));
+  EXPECT_EQ(sink.FindMatch(true, "foo/baz"), LoggingSink::Config(512, 2048));
+  // server test
+  EXPECT_EQ(sink.FindMatch(false, "foo/bar"), LoggingSink::Config(0, 0));
+  EXPECT_EQ(sink.FindMatch(false, "foo/baz"), LoggingSink::Config(0, 0));
+}
+
+TEST(GcpObservabilityLoggingSinkTest, LoggingConfigServerWildCardEntries) {
+  const char* json_str = R"json({
+      "cloud_logging": {
+        "server_rpc_events": [
+          {
+            "methods": ["*"],
+            "max_metadata_bytes": 1024,
+            "max_message_bytes": 4096
+          }
+        ]
+      }
+    })json";
+  auto json = grpc_core::Json::Parse(json_str);
+  ASSERT_TRUE(json.ok()) << json.status();
+  grpc_core::ValidationErrors errors;
+  auto config = grpc_core::LoadFromJson<GcpObservabilityConfig>(
+      *json, grpc_core::JsonArgs(), &errors);
+  ASSERT_TRUE(errors.ok()) << errors.status("unexpected errors");
+  ObservabilityLoggingSink sink(config.cloud_logging.value());
+  // client test
+  EXPECT_EQ(sink.FindMatch(true, "foo/bar"), LoggingSink::Config(0, 0));
+  // server test
+  EXPECT_EQ(sink.FindMatch(false, "foo/bar"), LoggingSink::Config(1024, 4096));
+}
+
+TEST(GcpObservabilityLoggingSinkTest,
+     LoggingConfigServerWildCardServiceEntries) {
+  const char* json_str = R"json({
+      "cloud_logging": {
+        "server_rpc_events": [
+          {
+            "methods": ["service/*"],
+            "max_metadata_bytes": 1024,
+            "max_message_bytes": 4096
+          }
+        ]
+      }
+    })json";
+  auto json = grpc_core::Json::Parse(json_str);
+  ASSERT_TRUE(json.ok()) << json.status();
+  grpc_core::ValidationErrors errors;
+  auto config = grpc_core::LoadFromJson<GcpObservabilityConfig>(
+      *json, grpc_core::JsonArgs(), &errors);
+  ASSERT_TRUE(errors.ok()) << errors.status("unexpected errors");
+  ObservabilityLoggingSink sink(config.cloud_logging.value());
+  // client test
+  EXPECT_EQ(sink.FindMatch(true, "service/bar"), LoggingSink::Config(0, 0));
+  EXPECT_EQ(sink.FindMatch(true, "foo/bar"), LoggingSink::Config(0, 0));
+  // server test
+  EXPECT_EQ(sink.FindMatch(false, "service/bar"),
+            LoggingSink::Config(1024, 4096));
+  EXPECT_EQ(sink.FindMatch(false, "foo/bar"), LoggingSink::Config(0, 0));
+}
+
+TEST(GcpObservabilityLoggingSinkTest,
+     LoggingConfigServerMultipleMethodEntries) {
+  const char* json_str = R"json({
+      "cloud_logging": {
+        "server_rpc_events": [
+          {
+            "methods": ["foo/bar", "foo/baz"],
+            "max_metadata_bytes": 1024,
+            "max_message_bytes": 4096
+          }
+        ]
+      }
+    })json";
+  auto json = grpc_core::Json::Parse(json_str);
+  ASSERT_TRUE(json.ok()) << json.status();
+  grpc_core::ValidationErrors errors;
+  auto config = grpc_core::LoadFromJson<GcpObservabilityConfig>(
+      *json, grpc_core::JsonArgs(), &errors);
+  ASSERT_TRUE(errors.ok()) << errors.status("unexpected errors");
+  ObservabilityLoggingSink sink(config.cloud_logging.value());
+  // client test
+  EXPECT_EQ(sink.FindMatch(true, "foo/bar"), LoggingSink::Config(0, 0));
+  EXPECT_EQ(sink.FindMatch(true, "foo/baz"), LoggingSink::Config(0, 0));
+  // server test
+  EXPECT_EQ(sink.FindMatch(false, "foo/bar"), LoggingSink::Config(1024, 4096));
+  EXPECT_EQ(sink.FindMatch(false, "foo/baz"), LoggingSink::Config(1024, 4096));
+}
+
+TEST(GcpObservabilityLoggingSinkTest, LoggingConfigServerMultipleEventEntries) {
+  const char* json_str = R"json({
+      "cloud_logging": {
+        "server_rpc_events": [
+          {
+            "methods": ["foo/bar"],
+            "max_metadata_bytes": 1024,
+            "max_message_bytes": 4096
+          },
+          {
+            "methods": ["foo/baz"],
+            "max_metadata_bytes": 512,
+            "max_message_bytes": 2048
+          }
+        ]
+      }
+    })json";
+  auto json = grpc_core::Json::Parse(json_str);
+  ASSERT_TRUE(json.ok()) << json.status();
+  grpc_core::ValidationErrors errors;
+  auto config = grpc_core::LoadFromJson<GcpObservabilityConfig>(
+      *json, grpc_core::JsonArgs(), &errors);
+  ASSERT_TRUE(errors.ok()) << errors.status("unexpected errors");
+  ObservabilityLoggingSink sink(config.cloud_logging.value());
+  // client test
+  EXPECT_EQ(sink.FindMatch(true, "foo/bar"), LoggingSink::Config(0, 0));
+  EXPECT_EQ(sink.FindMatch(true, "foo/baz"), LoggingSink::Config(0, 0));
+  // server test
+  EXPECT_EQ(sink.FindMatch(false, "foo/bar"), LoggingSink::Config(1024, 4096));
+  EXPECT_EQ(sink.FindMatch(false, "foo/baz"), LoggingSink::Config(512, 2048));
+}
+
+}  // namespace
+
+}  // namespace internal
+}  // namespace grpc
+
+int main(int argc, char** argv) {
+  grpc::testing::TestEnvironment env(&argc, argv);
+  ::testing::InitGoogleTest(&argc, argv);
+  return RUN_ALL_TESTS();
+}