[release-1.4] Add extra labels to sd logging

extensions/common/context.cc

@@ -289,6 +289,29 @@ void populateHTTPRequestInfo(bool outbound, bool use_host_header_fallback,
   getValue({"response", "total_size"}, &request_info->response_size);
 }
 
+void populateExtendedHTTPRequestInfo(RequestInfo* request_info) {
+  getStringValue({"source", "address"}, &request_info->source_address);
+  getStringValue({"destination", "address"},
+                 &request_info->destination_address);
+
+  getStringValue({"request", "referer"}, &request_info->referer);
+  getStringValue({"request", "user_agent"}, &request_info->user_agent);
+  getStringValue({"request", "id"}, &request_info->request_id);
+  std::string trace_sampled;
+  if (getStringValue({"request", "headers", "x-b3-sampled"}, &trace_sampled) &&
+      trace_sampled == "1") {
+    getStringValue({"request", "headers", "x-b3-traceid"},
+                   &request_info->b3_trace_id);
+    getStringValue({"request", "headers", "x-b3-spanid"},
+                   &request_info->b3_span_id);
+    request_info->b3_trace_sampled = true;
+  }
+
+  getStringValue({"request", "url_path"}, &request_info->url_path);
+  getStringValue({"request", "host"}, &request_info->url_host);
+  getStringValue({"request", "scheme"}, &request_info->url_scheme);
+}
+
 google::protobuf::util::Status extractNodeMetadataValue(
     const google::protobuf::Struct& node_metadata,
     google::protobuf::Struct* metadata) {

extensions/common/context.h

@@ -114,6 +114,24 @@ struct RequestInfo {
   // Rbac filter policy id and result.
   std::string rbac_permissive_policy_id;
   std::string rbac_permissive_engine_result;
+
+  // The following fields will only be populated by calling
+  // populateExtendedHTTPRequestInfo.
+  std::string source_address;
+  std::string destination_address;
+
+  // Important Headers.
+  std::string referer;
+  std::string user_agent;
+  std::string request_id;
+  std::string b3_trace_id;
+  std::string b3_span_id;
+  bool b3_trace_sampled = false;
+
+  // HTTP URL related attributes.
+  std::string url_path;
+  std::string url_host;
+  std::string url_scheme;
 };
 
 // RequestContext contains all the information available in the request.
@@ -156,6 +174,10 @@ void populateHTTPRequestInfo(bool outbound, bool use_host_header,
                              RequestInfo* request_info,
                              const std::string& destination_namespace);
 
+// populateExtendedHTTPRequestInfo populates the extra fields in RequestInfo
+// struct, includes trace headers, request id headers, and url.
+void populateExtendedHTTPRequestInfo(RequestInfo* request_info);
+
 // Extracts node metadata value. It looks for values of all the keys
 // corresponding to EXCHANGE_KEYS in node_metadata and populates it in
 // google::protobuf::Value pointer that is passed in.

extensions/stackdriver/log/logger.cc

@@ -46,11 +46,10 @@ Logger::Logger(const ::wasm::common::NodeInfo& local_node_info,
   // Set log names.
   const auto& platform_metadata = local_node_info.platform_metadata();
   const auto project_iter = platform_metadata.find(Common::kGCPProjectKey);
-  std::string project_id = "";
   if (project_iter != platform_metadata.end()) {
-    project_id = project_iter->second;
+    project_id_ = project_iter->second;
   }
-  log_entries_request_->set_log_name("projects/" + project_id + "/logs/" +
+  log_entries_request_->set_log_name("projects/" + project_id_ + "/logs/" +
                                      kServerAccessLogName);
 
   std::string resource_type = Common::kContainerMonitoredResource;
@@ -72,6 +71,17 @@ Logger::Logger(const ::wasm::common::NodeInfo& local_node_info,
   (*label_map)["destination_workload"] = local_node_info.workload_name();
   (*label_map)["destination_namespace"] = local_node_info.namespace_();
   (*label_map)["mesh_uid"] = local_node_info.mesh_id();
+  // Add destination app and version label if exist.
+  const auto& local_labels = local_node_info.labels();
+  auto version_iter = local_labels.find("version");
+  if (version_iter != local_labels.end()) {
+    (*label_map)["destination_version"] = version_iter->second;
+  }
+  // App label is used to correlate workload and its logs in UI.
+  auto app_iter = local_labels.find("app");
+  if (app_iter != local_labels.end()) {
+    (*label_map)["destination_app"] = app_iter->second;
+  }
   log_request_size_limit_ = log_request_size_limit;
   exporter_ = std::move(exporter);
 }
@@ -89,20 +99,57 @@ void Logger::addLogEntry(const ::Wasm::Common::RequestInfo& request_info,
       absl::Nanoseconds(1));
   new_entry->set_severity(::google::logging::type::INFO);
   auto label_map = new_entry->mutable_labels();
+  (*label_map)["request_id"] = request_info.request_id;
   (*label_map)["source_name"] = peer_node_info.name();
   (*label_map)["source_workload"] = peer_node_info.workload_name();
   (*label_map)["source_namespace"] = peer_node_info.namespace_();
+  // Add source app and version label if exist.
+  const auto& peer_labels = peer_node_info.labels();
+  auto version_iter = peer_labels.find("version");
+  if (version_iter != peer_labels.end()) {
+    (*label_map)["source_version"] = version_iter->second;
+  }
+  auto app_iter = peer_labels.find("app");
+  if (app_iter != peer_labels.end()) {
+    (*label_map)["source_app"] = app_iter->second;
+  }
 
-  (*label_map)["request_operation"] = request_info.request_operation;
   (*label_map)["destination_service_host"] =
       request_info.destination_service_host;
   (*label_map)["response_flag"] = request_info.response_flag;
-  (*label_map)["protocol"] = request_info.request_protocol;
   (*label_map)["destination_principal"] = request_info.destination_principal;
   (*label_map)["source_principal"] = request_info.source_principal;
   (*label_map)["service_authentication_policy"] =
       std::string(::Wasm::Common::AuthenticationPolicyString(
           request_info.service_auth_policy));
+
+  // Insert HTTPRequest
+  auto http_request = new_entry->mutable_http_request();
+  http_request->set_request_method(request_info.request_operation);
+  http_request->set_request_url(request_info.url_scheme + "://" +
+                                request_info.url_host + request_info.url_path);
+  http_request->set_request_size(request_info.request_size);
+  http_request->set_status(request_info.response_code);
+  http_request->set_response_size(request_info.response_size);
+  http_request->set_user_agent(request_info.user_agent);
+  http_request->set_remote_ip(request_info.source_address);
+  http_request->set_server_ip(request_info.destination_address);
+  http_request->set_protocol(request_info.request_protocol);
+  auto duration = request_info.duration;
+  http_request->mutable_latency()->set_seconds(
+      absl::IDivDuration(duration, absl::Seconds(1), &duration));
+  http_request->mutable_latency()->set_nanos(
+      absl::IDivDuration(duration, absl::Nanoseconds(1), &duration));
+  http_request->set_referer(request_info.referer);
+
+  // Insert trace headers, if exist.
+  if (request_info.b3_trace_sampled) {
+    new_entry->set_trace("projects/" + project_id_ + "/traces/" +
+                         request_info.b3_trace_id);
+    new_entry->set_span_id(request_info.b3_span_id);
+    new_entry->set_trace_sampled(request_info.b3_trace_sampled);
+  }
+
   // Accumulate estimated size of the request. If the current request exceeds
   // the size limit, flush the request out.
   size_ += new_entry->ByteSizeLong();

extensions/stackdriver/log/logger.h

@@ -74,6 +74,9 @@ class Logger {
 
   // Exporter calls Stackdriver services to export access logs.
   std::unique_ptr<Exporter> exporter_;
+
+  // GCP project that this proxy runs with.
+  std::string project_id_;
 };
 
 }  // namespace Log

extensions/stackdriver/log/logger_test.cc

@@ -21,6 +21,7 @@
 #include "extensions/stackdriver/common/utils.h"
 #include "gmock/gmock.h"
 #include "google/logging/v2/log_entry.pb.h"
+#include "google/protobuf/util/json_util.h"
 #include "google/protobuf/util/message_differencer.h"
 #include "gtest/gtest.h"
 
@@ -31,9 +32,6 @@ namespace Log {
 using google::protobuf::util::MessageDifferencer;
 using google::protobuf::util::TimeUtil;
 
-constexpr char kServerAccessLogName[] =
-    "projects/test_project/logs/server-accesslog-stackdriver";
-
 namespace {
 
 class MockExporter : public Exporter {
@@ -68,6 +66,7 @@ wasm::common::NodeInfo peerNodeInfo() {
   (*node_info.mutable_platform_metadata())[Common::kGCPLocationKey] =
       "test_location";
   node_info.set_namespace_("test_peer_namespace");
+  node_info.set_workload_name("test_peer_workload");
   node_info.set_name("test_peer_pod");
   return node_info;
 }
@@ -83,44 +82,80 @@ ::Wasm::Common::RequestInfo requestInfo() {
   request_info.source_principal = "source_principal";
   request_info.service_auth_policy =
       ::Wasm::Common::ServiceAuthenticationPolicy::MutualTLS;
+  request_info.duration = absl::Seconds(10);  // 10s
+  request_info.url_scheme = "http";
+  request_info.url_host = "httpbin.org";
+  request_info.url_path = "/headers";
+  request_info.request_id = "123";
+  request_info.b3_trace_id = "123abc";
+  request_info.b3_span_id = "abc123";
+  request_info.b3_trace_sampled = true;
+  request_info.user_agent = "chrome";
+  request_info.referer = "www.google.com";
+  request_info.source_address = "1.1.1.1";
+  request_info.destination_address = "2.2.2.2";
   return request_info;
 }
 
+std::string write_log_request_json = R"({ 
+  "logName":"projects/test_project/logs/server-accesslog-stackdriver",
+  "resource":{ 
+     "type":"k8s_container",
+     "labels":{ 
+        "cluster_name":"test_cluster",
+        "pod_name":"test_pod",
+        "location":"test_location",
+        "namespace_name":"test_namespace",
+        "project_id":"test_project",
+        "container_name":"istio-proxy"
+     }
+  },
+  "labels":{ 
+     "destination_workload":"test_workload",
+     "mesh_uid":"mesh",
+     "destination_namespace":"test_namespace",
+     "destination_name":"test_pod"
+  },
+  "entries":[ 
+     {
+        "httpRequest":{ 
+           "requestMethod":"GET",
+           "requestUrl":"http://httpbin.org/headers",
+           "userAgent":"chrome",
+           "remoteIp":"1.1.1.1",
+           "referer":"www.google.com",
+           "serverIp":"2.2.2.2",
+           "latency":"10s",
+           "protocol":"HTTP"
+        },
+        "timestamp":"1970-01-01T00:00:00Z",
+        "severity":"INFO",
+        "labels":{ 
+           "source_name":"test_peer_pod",
+           "destination_principal":"destination_principal",
+           "destination_service_host":"httpbin.org",
+           "request_id":"123",
+           "source_namespace":"test_peer_namespace",
+           "source_principal":"source_principal",
+           "service_authentication_policy":"MUTUAL_TLS",
+           "source_workload":"test_peer_workload",
+           "response_flag":"-"
+        },
+        "trace":"projects/test_project/traces/123abc",
+        "spanId":"abc123",
+        "traceSampled":true
+     }
+  ]
+})";
+
 google::logging::v2::WriteLogEntriesRequest expectedRequest(
     int log_entry_count) {
-  auto request_info = requestInfo();
-  auto peer_node_info = peerNodeInfo();
-  auto node_info = nodeInfo();
   google::logging::v2::WriteLogEntriesRequest req;
-  req.set_log_name(kServerAccessLogName);
-  google::api::MonitoredResource monitored_resource;
-  Common::getMonitoredResource(Common::kContainerMonitoredResource, node_info,
-                               &monitored_resource);
-  req.mutable_resource()->CopyFrom(monitored_resource);
-  auto top_label_map = req.mutable_labels();
-  (*top_label_map)["destination_name"] = node_info.name();
-  (*top_label_map)["destination_workload"] = node_info.workload_name();
-  (*top_label_map)["destination_namespace"] = node_info.namespace_();
-  (*top_label_map)["mesh_uid"] = node_info.mesh_id();
-  for (int i = 0; i < log_entry_count; i++) {
+  google::protobuf::util::JsonParseOptions options;
+  JsonStringToMessage(write_log_request_json, &req, options);
+  for (int i = 1; i < log_entry_count; i++) {
     auto* new_entry = req.mutable_entries()->Add();
-    *new_entry->mutable_timestamp() = TimeUtil::SecondsToTimestamp(0);
-    new_entry->set_severity(::google::logging::type::INFO);
-    auto label_map = new_entry->mutable_labels();
-    (*label_map)["source_name"] = peer_node_info.name();
-    (*label_map)["source_workload"] = peer_node_info.workload_name();
-    (*label_map)["source_namespace"] = peer_node_info.namespace_();
-
-    (*label_map)["request_operation"] = request_info.request_operation;
-    (*label_map)["destination_service_host"] =
-        request_info.destination_service_host;
-    (*label_map)["response_flag"] = request_info.response_flag;
-    (*label_map)["protocol"] = request_info.request_protocol;
-    (*label_map)["destination_principal"] = request_info.destination_principal;
-    (*label_map)["source_principal"] = request_info.source_principal;
-    (*label_map)["service_authentication_policy"] =
-        std::string(::Wasm::Common::AuthenticationPolicyString(
-            request_info.service_auth_policy));
+    new_entry->CopyFrom(req.entries()[0]);
   }
   return req;
 }
@@ -137,9 +172,13 @@ TEST(LoggerTest, TestWriteLogEntry) {
           [](const std::vector<std::unique_ptr<
                  const google::logging::v2::WriteLogEntriesRequest>>&
                  requests) {
-            auto expected_request = expectedRequest(1);
             for (const auto& req : requests) {
-              EXPECT_TRUE(MessageDifferencer::Equals(expected_request, *req));
+              std::string diff;
+              MessageDifferencer differ;
+              differ.ReportDifferencesToString(&diff);
+              if (!differ.Compare(expectedRequest(1), *req)) {
+                FAIL() << "unexpected log entry " << diff << "\n";
+              }
             }
           }));
   logger->exportLogEntry();
@@ -148,7 +187,7 @@ TEST(LoggerTest, TestWriteLogEntry) {
 TEST(LoggerTest, TestWriteLogEntryRotation) {
   auto exporter = std::make_unique<::testing::NiceMock<MockExporter>>();
   auto exporter_ptr = exporter.get();
-  auto logger = std::make_unique<Logger>(nodeInfo(), std::move(exporter), 900);
+  auto logger = std::make_unique<Logger>(nodeInfo(), std::move(exporter), 1200);
   for (int i = 0; i < 9; i++) {
     logger->addLogEntry(requestInfo(), peerNodeInfo());
   }
@@ -159,8 +198,12 @@ TEST(LoggerTest, TestWriteLogEntryRotation) {
                  requests) {
             EXPECT_EQ(requests.size(), 3);
             for (const auto& req : requests) {
-              auto expected_request = expectedRequest(3);
-              EXPECT_TRUE(MessageDifferencer::Equals(expected_request, *req));
+              std::string diff;
+              MessageDifferencer differ;
+              differ.ReportDifferencesToString(&diff);
+              if (!differ.Compare(expectedRequest(3), *req)) {
+                FAIL() << "unexpected log entry " << diff << "\n";
+              }
             }
           }));
   logger->exportLogEntry();

extensions/stackdriver/stackdriver.cc

@@ -214,6 +214,7 @@ void StackdriverRootContext::record() {
   ::Extensions::Stackdriver::Metric::record(isOutbound(), local_node_info_,
                                             peer_node_info, request_info);
   if (enableServerAccessLog()) {
+    ::Wasm::Common::populateExtendedHTTPRequestInfo(&request_info);
     logger_->addLogEntry(request_info, peer_node_info);
   }
   if (enableEdgeReporting()) {

test/envoye2e/driver/stackdriver.go

@@ -65,9 +65,14 @@ func (sd *Stackdriver) Run(p *Params) error {
 				sd.Unlock()
 			case req := <-logging.RcvLoggingReq:
 				log.Println("sd received log request")
-				// clear the timestamps for comparison
+				// clear the timestamps, latency request id, and req/resp size for comparison
 				for _, entry := range req.Entries {
 					entry.Timestamp = nil
+					entry.HttpRequest.RequestSize = 0
+					entry.HttpRequest.ResponseSize = 0
+					entry.HttpRequest.Latency = nil
+					entry.HttpRequest.RemoteIp = ""
+					delete(entry.Labels, "request_id")
 				}
 				sd.Lock()
 				sd.ls[proto.MarshalTextString(req)] = struct{}{}

test/envoye2e/stackdriver_plugin/stackdriver_plugin_test.go

@@ -88,6 +88,11 @@ func compareTimeSeries(got, want *monitoringpb.TimeSeries) error {
 func compareLogEntries(got, want *logging.WriteLogEntriesRequest) error {
 	for _, l := range got.Entries {
 		l.Timestamp = nil
+		delete(l.Labels, "request_id")
+		l.HttpRequest.RequestSize = 0
+		l.HttpRequest.ResponseSize = 0
+		l.HttpRequest.Latency = nil
+		l.HttpRequest.RemoteIp = ""
 	}
 	if !proto.Equal(want, got) {
 		return fmt.Errorf("log entries are not expected, got %v \nwant %v\n", proto.MarshalTextString(got), proto.MarshalTextString(want))
@@ -124,7 +129,10 @@ func verifyWriteLogEntriesReq(got *logging.WriteLogEntriesRequest) error {
 	var srvLogReq logging.WriteLogEntriesRequest
 	p := &driver.Params{
 		Vars: map[string]string{
+			"ServerPort":                  "20045",
+			"ClientPort":                  "20042",
 			"ServiceAuthenticationPolicy": "NONE",
+			"RequestPath":                 "echo",
 		},
 	}
 	p.LoadTestProto("testdata/stackdriver/server_access_log.yaml.tmpl", &srvLogReq)