Perfetto: Add Chrome tracing metadata

This adds support for both global metadata, and metadata events.

As part of this, the Perfetto support expands in a couple of ways:
* Implement Flush(), so a Consumer can make sure the metadata events
  are written once, before tracing shuts down.
* Support multiple data sources; global metadata is a separate
  data source as it's not actually tied to tracing.

Bug: 840499
Change-Id: Ib4a76f692f333ffea89f7cf8200d0da3eeefaa23
Reviewed-on: https://chromium-review.googlesource.com/1129959
Commit-Queue: oysteine <oysteine@chromium.org>
Reviewed-by: Daniel Cheng <dcheng@chromium.org>
Reviewed-by: Sami Kyöstilä <skyostil@chromium.org>
Reviewed-by: Primiano Tucci <primiano@chromium.org>
Cr-Commit-Position: refs/heads/master@{#580454}

base/trace_event/trace_log.cc

@@ -325,9 +325,13 @@ void TraceLog::ThreadLocalEventBuffer::FlushWhileLocked() {
 }
 
 void TraceLog::SetAddTraceEventOverride(
-    const AddTraceEventOverrideCallback& override) {
+    const AddTraceEventOverrideCallback& override,
+    const OnFlushCallback& on_flush_callback) {
   subtle::NoBarrier_Store(&trace_event_override_,
                           reinterpret_cast<subtle::AtomicWord>(override));
+  subtle::NoBarrier_Store(
+      &on_flush_callback_,
+      reinterpret_cast<subtle::AtomicWord>(on_flush_callback));
 }
 
 struct TraceLog::RegisteredAsyncObserver {
@@ -371,6 +375,7 @@ TraceLog::TraceLog()
       generation_(0),
       use_worker_thread_(false),
       trace_event_override_(0),
+      on_flush_callback_(0),
       filter_factory_for_testing_(nullptr) {
   CategoryRegistry::Initialize();
 
@@ -991,6 +996,12 @@ void TraceLog::FlushCurrentThread(int generation, bool discard_events) {
   // This will flush the thread local buffer.
   delete thread_local_event_buffer_.Get();
 
+  auto on_flush_callback = reinterpret_cast<OnFlushCallback>(
+      subtle::NoBarrier_Load(&on_flush_callback_));
+  if (on_flush_callback) {
+    on_flush_callback();
+  }
+
   // Scheduler uses TRACE_EVENT macros when posting a task, which can lead
   // to acquiring a tracing lock. Given that posting a task requires grabbing
   // a scheduler lock, we need to post this task outside tracing lock to avoid
@@ -1523,45 +1534,69 @@ uint64_t TraceLog::MangleEventId(uint64_t id) {
   return id ^ process_id_hash_;
 }
 
+template <typename T>
+void TraceLog::AddMetadataEventWhileLocked(int thread_id,
+                                           const char* metadata_name,
+                                           const char* arg_name,
+                                           const T& value) {
+  auto trace_event_override = reinterpret_cast<AddTraceEventOverrideCallback>(
+      subtle::NoBarrier_Load(&trace_event_override_));
+  if (trace_event_override) {
+    TraceEvent trace_event;
+    InitializeMetadataEvent(&trace_event, thread_id, metadata_name, arg_name,
+                            value);
+    trace_event_override(trace_event);
+  } else {
+    InitializeMetadataEvent(
+        AddEventToThreadSharedChunkWhileLocked(nullptr, false), thread_id,
+        metadata_name, arg_name, value);
+  }
+}
+
 void TraceLog::AddMetadataEventsWhileLocked() {
   lock_.AssertAcquired();
 
+  auto trace_event_override = reinterpret_cast<AddTraceEventOverrideCallback>(
+      subtle::NoBarrier_Load(&trace_event_override_));
+
   // Move metadata added by |AddMetadataEvent| into the trace log.
+  if (trace_event_override) {
+    while (!metadata_events_.empty()) {
+      trace_event_override(*metadata_events_.back());
+      metadata_events_.pop_back();
+    }
+  } else {
     while (!metadata_events_.empty()) {
-    TraceEvent* event = AddEventToThreadSharedChunkWhileLocked(nullptr, false);
+      TraceEvent* event =
+          AddEventToThreadSharedChunkWhileLocked(nullptr, false);
       event->MoveFrom(std::move(metadata_events_.back()));
       metadata_events_.pop_back();
     }
+  }
 
 #if !defined(OS_NACL)  // NaCl shouldn't expose the process id.
-  InitializeMetadataEvent(
-      AddEventToThreadSharedChunkWhileLocked(nullptr, false), 0, "num_cpus",
-      "number", base::SysInfo::NumberOfProcessors());
+  AddMetadataEventWhileLocked(0, "num_cpus", "number",
+                              base::SysInfo::NumberOfProcessors());
 #endif
 
   int current_thread_id = static_cast<int>(base::PlatformThread::CurrentId());
   if (process_sort_index_ != 0) {
-    InitializeMetadataEvent(
-        AddEventToThreadSharedChunkWhileLocked(nullptr, false),
-        current_thread_id, "process_sort_index", "sort_index",
-        process_sort_index_);
+    AddMetadataEventWhileLocked(current_thread_id, "process_sort_index",
+                                "sort_index", process_sort_index_);
   }
 
   if (!process_name_.empty()) {
-    InitializeMetadataEvent(
-        AddEventToThreadSharedChunkWhileLocked(nullptr, false),
-        current_thread_id, "process_name", "name", process_name_);
+    AddMetadataEventWhileLocked(current_thread_id, "process_name", "name",
+                                process_name_);
   }
 
   TimeDelta process_uptime = TRACE_TIME_NOW() - process_creation_time_;
-  InitializeMetadataEvent(
-      AddEventToThreadSharedChunkWhileLocked(nullptr, false), current_thread_id,
-      "process_uptime_seconds", "uptime", process_uptime.InSeconds());
+  AddMetadataEventWhileLocked(current_thread_id, "process_uptime_seconds",
+                              "uptime", process_uptime.InSeconds());
 
 #if defined(OS_ANDROID)
-  InitializeMetadataEvent(
-      AddEventToThreadSharedChunkWhileLocked(nullptr, false), current_thread_id,
-      "chrome_library_address", "start_address",
+  AddMetadataEventWhileLocked(current_thread_id, "chrome_library_address",
+                              "start_address",
                               base::StringPrintf("%p", &__executable_start));
 #endif
 
@@ -1569,9 +1604,7 @@ void TraceLog::AddMetadataEventsWhileLocked() {
     std::vector<base::StringPiece> labels;
     for (const auto& it : process_labels_)
       labels.push_back(it.second);
-    InitializeMetadataEvent(
-        AddEventToThreadSharedChunkWhileLocked(nullptr, false),
-        current_thread_id, "process_labels", "labels",
+    AddMetadataEventWhileLocked(current_thread_id, "process_labels", "labels",
                                 base::JoinString(labels, ","));
   }
 
@@ -1579,9 +1612,8 @@ void TraceLog::AddMetadataEventsWhileLocked() {
   for (const auto& it : thread_sort_indices_) {
     if (it.second == 0)
       continue;
-    InitializeMetadataEvent(
-        AddEventToThreadSharedChunkWhileLocked(nullptr, false), it.first,
-        "thread_sort_index", "sort_index", it.second);
+    AddMetadataEventWhileLocked(it.first, "thread_sort_index", "sort_index",
+                                it.second);
   }
 
   // Thread names.
@@ -1589,16 +1621,13 @@ void TraceLog::AddMetadataEventsWhileLocked() {
   for (const auto& it : thread_names_) {
     if (it.second.empty())
       continue;
-    InitializeMetadataEvent(
-        AddEventToThreadSharedChunkWhileLocked(nullptr, false), it.first,
-        "thread_name", "name", it.second);
+    AddMetadataEventWhileLocked(it.first, "thread_name", "name", it.second);
   }
 
   // If buffer is full, add a metadata record to report this.
   if (!buffer_limit_reached_timestamp_.is_null()) {
-    InitializeMetadataEvent(
-        AddEventToThreadSharedChunkWhileLocked(nullptr, false),
-        current_thread_id, "trace_buffer_overflowed", "overflowed_at_ts",
+    AddMetadataEventWhileLocked(current_thread_id, "trace_buffer_overflowed",
+                                "overflowed_at_ts",
                                 buffer_limit_reached_timestamp_);
   }
 }

base/trace_event/trace_log.h

@@ -179,10 +179,12 @@ class BASE_EXPORT TraceLog : public MemoryDumpProvider {
   void CancelTracing(const OutputCallback& cb);
 
   typedef void (*AddTraceEventOverrideCallback)(const TraceEvent&);
+  typedef void (*OnFlushCallback)();
   // The callback will be called up until the point where the flush is
   // finished, i.e. must be callable until OutputCallback is called with
   // has_more_events==false.
-  void SetAddTraceEventOverride(const AddTraceEventOverrideCallback& override);
+  void SetAddTraceEventOverride(const AddTraceEventOverrideCallback& override,
+                                const OnFlushCallback& on_flush_callback);
 
   // Called by TRACE_EVENT* macros, don't call this directly.
   // The name parameter is a category group for example:
@@ -394,6 +396,11 @@ class BASE_EXPORT TraceLog : public MemoryDumpProvider {
   TraceLog();
   ~TraceLog() override;
   void AddMetadataEventsWhileLocked();
+  template <typename T>
+  void AddMetadataEventWhileLocked(int thread_id,
+                                   const char* metadata_name,
+                                   const char* arg_name,
+                                   const T& value);
 
   InternalTraceOptions trace_options() const {
     return static_cast<InternalTraceOptions>(
@@ -516,6 +523,7 @@ class BASE_EXPORT TraceLog : public MemoryDumpProvider {
   subtle::AtomicWord generation_;
   bool use_worker_thread_;
   subtle::AtomicWord trace_event_override_;
+  subtle::AtomicWord on_flush_callback_;
 
   FilterFactoryForTesting filter_factory_for_testing_;
 

content/browser/tracing/background_tracing_manager_impl.cc

@@ -561,6 +561,8 @@ bool BackgroundTracingManagerImpl::IsAllowedFinalization() const {
 
 std::unique_ptr<base::DictionaryValue>
 BackgroundTracingManagerImpl::GenerateMetadataDict() {
+  DCHECK_CURRENTLY_ON(BrowserThread::UI);
+
   auto metadata_dict = std::make_unique<base::DictionaryValue>();
   if (config_) {
     auto config_dict = std::make_unique<base::DictionaryValue>();

content/browser/tracing/tracing_controller_impl.cc

@@ -161,6 +161,7 @@ tracing::TraceEventAgent* TracingControllerImpl::GetTraceEventAgent() const {
 
 std::unique_ptr<base::DictionaryValue>
 TracingControllerImpl::GenerateMetadataDict() const {
+  DCHECK_CURRENTLY_ON(BrowserThread::UI);
   auto metadata_dict = std::make_unique<base::DictionaryValue>();
 
   // trace_config_ can be null if the tracing controller finishes flushing

services/tracing/perfetto/json_trace_exporter.cc

@@ -4,6 +4,10 @@
 
 #include "services/tracing/perfetto/json_trace_exporter.h"
 
+#include <utility>
+
+#include "base/json/json_reader.h"
+#include "base/json/json_writer.h"
 #include "base/json/string_escape.h"
 #include "base/logging.h"
 
@@ -178,7 +182,7 @@ namespace tracing {
 
 JSONTraceExporter::JSONTraceExporter(const std::string& config,
                                      perfetto::TracingService* service)
-    : config_(config) {
+    : config_(config), metadata_(std::make_unique<base::DictionaryValue>()) {
   consumer_endpoint_ = service->ConnectConsumer(this);
 }
 
@@ -188,23 +192,35 @@ void JSONTraceExporter::OnConnect() {
   // Start tracing.
   perfetto::TraceConfig trace_config;
   trace_config.add_buffers()->set_size_kb(4096 * 100);
-  auto* ds_config = trace_config.add_data_sources()->mutable_config();
-  ds_config->set_name(mojom::kTraceEventDataSourceName);
-  ds_config->set_target_buffer(0);
-  auto* chrome_config = ds_config->mutable_chrome_config();
+
+  auto* trace_event_config = trace_config.add_data_sources()->mutable_config();
+  trace_event_config->set_name(mojom::kTraceEventDataSourceName);
+  trace_event_config->set_target_buffer(0);
+  auto* chrome_config = trace_event_config->mutable_chrome_config();
   chrome_config->set_trace_config(config_);
 
+  auto* trace_metadata_config =
+      trace_config.add_data_sources()->mutable_config();
+  trace_metadata_config->set_name(mojom::kMetaDataSourceName);
+  trace_metadata_config->set_target_buffer(0);
+
   consumer_endpoint_->EnableTracing(trace_config);
 }
 
 void JSONTraceExporter::OnDisconnect() {}
 
+void JSONTraceExporter::OnTracingDisabled() {
+  consumer_endpoint_->ReadBuffers();
+}
+
+// This is called by the Coordinator interface, mainly used by the
+// TracingController which in turn is used by the tracing UI etc
+// to start/stop tracing.
 void JSONTraceExporter::StopAndFlush(OnTraceEventJSONCallback callback) {
   DCHECK(!json_callback_ && callback);
   json_callback_ = callback;
 
   consumer_endpoint_->DisableTracing();
-  consumer_endpoint_->ReadBuffers();
 }
 
 void JSONTraceExporter::OnTraceData(std::vector<perfetto::TracePacket> packets,
@@ -239,13 +255,37 @@ void JSONTraceExporter::OnTraceData(std::vector<perfetto::TracePacket> packets,
 
       OutputJSONFromTraceEventProto(event, &out);
     }
+
+    for (const perfetto::protos::ChromeMetadata& metadata : bundle.metadata()) {
+      if (metadata.has_string_value()) {
+        metadata_->SetString(metadata.name(), metadata.string_value());
+      } else if (metadata.has_int_value()) {
+        metadata_->SetInteger(metadata.name(), metadata.int_value());
+      } else if (metadata.has_bool_value()) {
+        metadata_->SetBoolean(metadata.name(), metadata.bool_value());
+      } else if (metadata.has_json_value()) {
+        std::unique_ptr<base::Value> value(
+            base::JSONReader::Read(metadata.json_value()));
+        metadata_->Set(metadata.name(), std::move(value));
+      } else {
+        NOTREACHED();
+      }
+    }
   }
 
   if (!has_more) {
-    out += "]}";
+    out += "]";
+    if (!metadata_->empty()) {
+      out += ",\"metadata\":";
+      std::string json_value;
+      base::JSONWriter::Write(*metadata_, &json_value);
+      out += json_value;
+    }
+
+    out += "}";
   }
 
-  json_callback_.Run(out, has_more);
+  json_callback_.Run(out, metadata_.get(), has_more);
 }
 
 }  // namespace tracing

services/tracing/perfetto/json_trace_exporter.h

@@ -11,6 +11,7 @@
 
 #include "base/callback.h"
 #include "base/macros.h"
+#include "base/values.h"
 
 #include "third_party/perfetto/include/perfetto/tracing/core/consumer.h"
 #include "third_party/perfetto/include/perfetto/tracing/core/tracing_service.h"
@@ -31,7 +32,9 @@ class JSONTraceExporter : public perfetto::Consumer {
   ~JSONTraceExporter() override;
 
   using OnTraceEventJSONCallback =
-      base::RepeatingCallback<void(const std::string& json, bool has_more)>;
+      base::RepeatingCallback<void(const std::string& json,
+                                   base::DictionaryValue* metadata,
+                                   bool has_more)>;
   void StopAndFlush(OnTraceEventJSONCallback callback);
 
   // perfetto::Consumer implementation.
@@ -39,7 +42,7 @@ class JSONTraceExporter : public perfetto::Consumer {
   // and to send finished protobufs over.
   void OnConnect() override;
   void OnDisconnect() override;
-  void OnTracingDisabled() override{};
+  void OnTracingDisabled() override;
   void OnTraceData(std::vector<perfetto::TracePacket> packets,
                    bool has_more) override;
 
@@ -48,6 +51,7 @@ class JSONTraceExporter : public perfetto::Consumer {
   bool has_output_json_preamble_ = false;
   bool has_output_first_event_ = false;
   std::string config_;
+  std::unique_ptr<base::DictionaryValue> metadata_;
 
   // Keep last to avoid edge-cases where its callbacks come in mid-destruction.
   std::unique_ptr<perfetto::TracingService::ConsumerEndpoint>

services/tracing/perfetto/json_trace_exporter_unittest.cc

@@ -15,6 +15,7 @@
 #include "base/trace_event/trace_event.h"
 #include "testing/gtest/include/gtest/gtest.h"
 
+#include "services/tracing/public/mojom/perfetto_service.mojom.h"
 #include "third_party/perfetto/include/perfetto/tracing/core/trace_config.h"
 #include "third_party/perfetto/include/perfetto/tracing/core/trace_packet.h"
 #include "third_party/perfetto/protos/perfetto/trace/chrome/chrome_trace_event.pbzero.h"
@@ -64,7 +65,8 @@ class MockConsumerEndpoint : public perfetto::TracingService::ConsumerEndpoint {
   void EnableTracing(
       const perfetto::TraceConfig& config,
       perfetto::base::ScopedFile = perfetto::base::ScopedFile()) override {
-    EXPECT_EQ(1, config.data_sources_size());
+    EXPECT_EQ(mojom::kTraceEventDataSourceName,
+              config.data_sources()[0].config().name());
     mock_service_->OnTracingEnabled(
         config.data_sources()[0].config().chrome_config().trace_config());
   }
@@ -72,7 +74,9 @@ class MockConsumerEndpoint : public perfetto::TracingService::ConsumerEndpoint {
   void DisableTracing() override { mock_service_->OnTracingDisabled(); }
   void ReadBuffers() override {}
   void FreeBuffers() override {}
-  void Flush(uint32_t timeout_ms, FlushCallback) override {}
+  void Flush(uint32_t timeout_ms, FlushCallback callback) override {
+    callback(true);
+  }
 
  private:
   MockService* mock_service_;
@@ -141,7 +145,10 @@ class JSONTraceExporterTest : public testing::Test {
         &JSONTraceExporterTest::OnTraceEventJSON, base::Unretained(this)));
   }
 
-  void OnTraceEventJSON(const std::string& json, bool has_more) {
+  void OnTraceEventJSON(const std::string& json,
+                        base::DictionaryValue* metadata,
+                        bool has_more) {
+    CHECK(!has_more);
     // The TraceAnalyzer expects the raw trace output, without the
     // wrapping root-node.
     static const size_t kTracingPreambleLength = strlen("\"{traceEvents\":");
@@ -151,7 +158,10 @@ class JSONTraceExporterTest : public testing::Test {
         json.length() - kTracingPreambleLength - kTracingEpilogueLength);
 
     trace_analyzer_.reset(trace_analyzer::TraceAnalyzer::Create(raw_events));
-    EXPECT_TRUE(trace_analyzer_);
+
+    parsed_trace_data_ =
+        base::DictionaryValue::From(base::JSONReader::Read(json));
+    EXPECT_TRUE(parsed_trace_data_);
   }
 
   void SetTestPacketBasicData(
@@ -215,12 +225,16 @@ class JSONTraceExporterTest : public testing::Test {
     return trace_analyzer_.get();
   }
   MockService* service() { return service_.get(); }
+  const base::DictionaryValue* parsed_trace_data() const {
+    return parsed_trace_data_.get();
+  }
 
  private:
   std::unique_ptr<MockService> service_;
   std::unique_ptr<JSONTraceExporter> json_trace_exporter_;
   std::unique_ptr<base::MessageLoop> message_loop_;
   std::unique_ptr<trace_analyzer::TraceAnalyzer> trace_analyzer_;
+  std::unique_ptr<base::DictionaryValue> parsed_trace_data_;
 };
 
 TEST_F(JSONTraceExporterTest, EnableTracingWithGivenConfig) {
@@ -230,6 +244,54 @@ TEST_F(JSONTraceExporterTest, EnableTracingWithGivenConfig) {
   EXPECT_EQ(kDummyTraceConfig, service()->tracing_enabled_with_config());
 }
 
+TEST_F(JSONTraceExporterTest, TestMetadata) {
+  CreateJSONTraceExporter("foo");
+  service()->WaitForTracingEnabled();
+  StopAndFlush();
+
+  perfetto::protos::TracePacket trace_packet_proto;
+
+  {
+    auto* new_metadata =
+        trace_packet_proto.mutable_chrome_events()->add_metadata();
+    new_metadata->set_name("int_metadata");
+    new_metadata->set_int_value(42);
+  }
+
+  {
+    auto* new_metadata =
+        trace_packet_proto.mutable_chrome_events()->add_metadata();
+    new_metadata->set_name("string_metadata");
+    new_metadata->set_string_value("met_val");
+  }
+
+  {
+    auto* new_metadata =
+        trace_packet_proto.mutable_chrome_events()->add_metadata();
+    new_metadata->set_name("bool_metadata");
+    new_metadata->set_bool_value(true);
+  }
+
+  {
+    auto* new_metadata =
+        trace_packet_proto.mutable_chrome_events()->add_metadata();
+    new_metadata->set_name("dict_metadata");
+    new_metadata->set_json_value("{\"child_dict\": \"foo\"}");
+  }
+
+  FinalizePacket(trace_packet_proto);
+
+  service()->WaitForTracingDisabled();
+  auto* metadata = parsed_trace_data()->FindKey("metadata");
+  EXPECT_TRUE(metadata);
+  EXPECT_EQ(metadata->FindKey("int_metadata")->GetInt(), 42);
+  EXPECT_EQ(metadata->FindKey("string_metadata")->GetString(), "met_val");
+  EXPECT_EQ(metadata->FindKey("bool_metadata")->GetBool(), true);
+  EXPECT_EQ(
+      metadata->FindKey("dict_metadata")->FindKey("child_dict")->GetString(),
+      "foo");
+}
+
 TEST_F(JSONTraceExporterTest, TestBasicEvent) {
   CreateJSONTraceExporter("foo");
   service()->WaitForTracingEnabled();

services/tracing/perfetto/perfetto_integration_unittest.cc

@@ -121,12 +121,16 @@ class MockProducerClient : public ProducerClient {
     }
   }
 
-  void TearDownDataSourceInstance(uint64_t id) override {
+  void TearDownDataSourceInstance(
+      uint64_t id,
+      TearDownDataSourceInstanceCallback callback) override {
     enabled_data_source_.reset();
 
     if (client_disabled_callback_) {
       std::move(client_disabled_callback_).Run();
     }
+
+    std::move(callback).Run();
   }
 
   void CommitData(const perfetto::CommitDataRequest& commit,

services/tracing/perfetto/perfetto_tracing_coordinator.cc

@@ -42,7 +42,9 @@ class PerfettoTracingCoordinator::TracingSession {
         &TracingSession::OnJSONTraceEventCallback, base::Unretained(this)));
   }
 
-  void OnJSONTraceEventCallback(const std::string& json, bool has_more) {
+  void OnJSONTraceEventCallback(const std::string& json,
+                                base::DictionaryValue* metadata,
+                                bool has_more) {
     if (stream_.is_valid()) {
       mojo::BlockingCopyFromString(json, stream_);
     }
@@ -50,7 +52,7 @@ class PerfettoTracingCoordinator::TracingSession {
     if (!has_more) {
       DCHECK(!stop_and_flush_callback_.is_null());
       base::ResetAndReturn(&stop_and_flush_callback_)
-          .Run(/*metadata=*/base::Value(base::Value::Type::DICTIONARY));
+          .Run(/*metadata=*/std::move(*metadata));
 
       base::SequencedTaskRunnerHandle::Get()->PostTask(
           FROM_HERE, std::move(tracing_over_callback_));

services/tracing/perfetto/producer_host.cc

@@ -53,11 +53,6 @@ void ProducerHost::OnConnectionError() {
 }
 
 void ProducerHost::OnConnect() {
-  // Register data sources with Perfetto here.
-
-  perfetto::DataSourceDescriptor descriptor;
-  descriptor.set_name(mojom::kTraceEventDataSourceName);
-  producer_endpoint_->RegisterDataSource(descriptor);
 }
 
 void ProducerHost::OnDisconnect() {
@@ -92,7 +87,13 @@ void ProducerHost::CreateDataSourceInstance(
 void ProducerHost::TearDownDataSourceInstance(
     perfetto::DataSourceInstanceID id) {
   if (producer_client_) {
-    producer_client_->TearDownDataSourceInstance(id);
+    producer_client_->TearDownDataSourceInstance(
+        id,
+        base::BindOnce(
+            [](ProducerHost* producer_host, perfetto::DataSourceInstanceID id) {
+              producer_host->producer_endpoint_->NotifyDataSourceStopped(id);
+            },
+            base::Unretained(this), id));
   }
 }
 
@@ -113,7 +114,10 @@ void ProducerHost::Flush(
 // inputs.
 void ProducerHost::CommitData(mojom::CommitDataRequestPtr data_request) {
   perfetto::CommitDataRequest native_data_request;
+
   // TODO(oysteine): Set up a TypeTrait for this instead of manual conversion.
+  native_data_request.set_flush_request_id(data_request->flush_request_id);
+
   for (auto& chunk : data_request->chunks_to_move) {
     auto* new_chunk = native_data_request.add_chunks_to_move();
     new_chunk->set_page(chunk->page);
@@ -145,4 +149,16 @@ void ProducerHost::CommitData(mojom::CommitDataRequestPtr data_request) {
   producer_endpoint_->CommitData(native_data_request);
 }
 
+void ProducerHost::RegisterDataSource(
+    mojom::DataSourceRegistrationPtr registration_info) {
+  perfetto::DataSourceDescriptor descriptor;
+  descriptor.set_name(registration_info->name);
+  descriptor.set_will_notify_on_stop(registration_info->will_notify_on_stop);
+  producer_endpoint_->RegisterDataSource(descriptor);
+}
+
+void ProducerHost::NotifyFlushComplete(uint64_t flush_request_id) {
+  producer_endpoint_->NotifyFlushComplete(flush_request_id);
+}
+
 }  // namespace tracing

services/tracing/perfetto/producer_host.h

@@ -69,12 +69,22 @@ class ProducerHost : public tracing::mojom::ProducerHost,
   void Flush(perfetto::FlushRequestID,
              const perfetto::DataSourceInstanceID* raw_data_source_ids,
              size_t num_data_sources) override;
+
   // mojom::ProducerHost implementation.
   // This interface gets called by the per-process ProducerClients
   // to signal that there's changes to be committed to the
   // Shared Memory buffer (like finished chunks).
   void CommitData(mojom::CommitDataRequestPtr data_request) override;
 
+  // Called by the ProducerClient to signal the Host that it can
+  // provide a specific data source.
+  void RegisterDataSource(
+      mojom::DataSourceRegistrationPtr registration_info) override;
+
+  // Called to signal the Host that a specific flush request
+  // is finished.
+  void NotifyFlushComplete(uint64_t flush_request_id) override;
+
  protected:
   void OnConnectionError();
 

services/tracing/public/cpp/perfetto/producer_client.cc

@@ -9,7 +9,6 @@
 #include "base/no_destructor.h"
 #include "base/task_scheduler/post_task.h"
 #include "services/tracing/public/cpp/perfetto/shared_memory.h"
-#include "services/tracing/public/cpp/perfetto/trace_event_data_source.h"
 #include "third_party/perfetto/include/perfetto/tracing/core/commit_data_request.h"
 #include "third_party/perfetto/include/perfetto/tracing/core/shared_memory_arbiter.h"
 #include "third_party/perfetto/include/perfetto/tracing/core/trace_writer.h"
@@ -33,7 +32,22 @@ PerfettoTaskRunner* GetPerfettoTaskRunner() {
 
 }  // namespace
 
-ProducerClient::ProducerClient() {
+ProducerClient::DataSourceBase::DataSourceBase(const std::string& name)
+    : name_(name) {
+  DCHECK(!name.empty());
+}
+
+ProducerClient::DataSourceBase::~DataSourceBase() = default;
+
+void ProducerClient::DataSourceBase::StartTracingWithID(
+    uint64_t data_source_id,
+    ProducerClient* producer_client,
+    const mojom::DataSourceConfig& data_source_config) {
+  data_source_id_ = data_source_id;
+  StartTracing(producer_client, data_source_config);
+}
+
+ProducerClient::ProducerClient() : weak_ptr_factory_(this) {
   DETACH_FROM_SEQUENCE(sequence_checker_);
 }
 
@@ -88,6 +102,21 @@ void ProducerClient::CreateMojoMessagepipesOnSequence(
                                 mojo::MakeRequest(&producer_host_)));
 }
 
+void ProducerClient::AddDataSource(DataSourceBase* data_source) {
+  GetTaskRunner()->PostTask(
+      FROM_HERE, base::BindOnce(&ProducerClient::AddDataSourceOnSequence,
+                                base::Unretained(this), data_source));
+}
+
+void ProducerClient::AddDataSourceOnSequence(DataSourceBase* data_source) {
+  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
+  data_sources_.insert(data_source);
+  auto new_registration = mojom::DataSourceRegistration::New();
+  new_registration->name = data_source->name();
+  new_registration->will_notify_on_stop = true;
+  producer_host_->RegisterDataSource(std::move(new_registration));
+}
+
 void ProducerClient::OnTracingStart(
     mojo::ScopedSharedBufferHandle shared_memory) {
   // TODO(oysteine): In next CLs plumb this through the service.
@@ -116,23 +145,47 @@ void ProducerClient::CreateDataSourceInstance(
   DCHECK(data_source_config);
 
   // TODO(oysteine): Support concurrent tracing sessions.
-  TraceEventDataSource::GetInstance()->StartTracing(this, *data_source_config);
+  for (auto* data_source : data_sources_) {
+    if (data_source->name() == data_source_config->name) {
+      data_source->StartTracingWithID(id, this, *data_source_config);
+      return;
+    }
+  }
 }
 
-void ProducerClient::TearDownDataSourceInstance(uint64_t id) {
+void ProducerClient::TearDownDataSourceInstance(
+    uint64_t id,
+    TearDownDataSourceInstanceCallback callback) {
   DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
 
-  TraceEventDataSource::GetInstance()->StopTracing();
+  for (auto* data_source : data_sources_) {
+    if (data_source->data_source_id() == id) {
+      data_source->StopTracing(std::move(callback));
+      return;
+    }
+  }
 
-  // TODO(oysteine): Yak shave: Can only destroy these once the TraceWriters
-  // are all cleaned up; have to figure out the TLS bits.
-  // shared_memory_arbiter_ = nullptr;
-  // shared_memory_ = nullptr;
+  LOG(FATAL) << "Invalid data source ID.";
 }
 
 void ProducerClient::Flush(uint64_t flush_request_id,
                            const std::vector<uint64_t>& data_source_ids) {
-  NOTREACHED();
+  pending_replies_for_latest_flush_ = {flush_request_id,
+                                       data_source_ids.size()};
+
+  // N^2, optimize once there's more than a couple of possible data sources.
+  for (auto* data_source : data_sources_) {
+    if (std::find(data_source_ids.begin(), data_source_ids.end(),
+                  data_source->data_source_id()) != data_source_ids.end()) {
+      data_source->Flush(base::BindRepeating(
+          [](base::WeakPtr<ProducerClient> weak_ptr, uint64_t id) {
+            if (weak_ptr) {
+              weak_ptr->NotifyFlushComplete(id);
+            }
+          },
+          weak_ptr_factory_.GetWeakPtr(), flush_request_id));
+    }
+  }
 }
 
 void ProducerClient::RegisterDataSource(const perfetto::DataSourceDescriptor&) {
@@ -143,6 +196,11 @@ void ProducerClient::UnregisterDataSource(const std::string& name) {
   NOTREACHED();
 }
 
+void ProducerClient::NotifyDataSourceStopped(
+    perfetto::DataSourceInstanceID id) {
+  NOTREACHED();
+}
+
 void ProducerClient::CommitData(const perfetto::CommitDataRequest& commit,
                                 CommitDataCallback callback) {
   // The CommitDataRequest which the SharedMemoryArbiter uses to
@@ -152,6 +210,7 @@ void ProducerClient::CommitData(const perfetto::CommitDataRequest& commit,
   // service-side.
   auto new_data_request = mojom::CommitDataRequest::New();
 
+  new_data_request->flush_request_id = commit.flush_request_id();
   for (auto& chunk : commit.chunks_to_move()) {
     auto new_chunk = mojom::ChunksToMove::New();
     new_chunk->page = chunk.page();
@@ -207,12 +266,17 @@ size_t ProducerClient::shared_buffer_page_size_kb() const {
   return 0;
 }
 
-void ProducerClient::NotifyFlushComplete(perfetto::FlushRequestID) {
-  NOTREACHED();
-}
+void ProducerClient::NotifyFlushComplete(perfetto::FlushRequestID id) {
+  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
+  if (pending_replies_for_latest_flush_.first != id) {
+    // Ignore; completed flush was for an earlier request.
+    return;
+  }
 
-void ProducerClient::NotifyDataSourceStopped(perfetto::DataSourceInstanceID) {
-  NOTREACHED();
+  DCHECK_NE(pending_replies_for_latest_flush_.second, 0u);
+  if (--pending_replies_for_latest_flush_.second == 0) {
+    producer_host_->NotifyFlushComplete(id);
+  }
 }
 
 std::unique_ptr<perfetto::TraceWriter> ProducerClient::CreateTraceWriter(

services/tracing/public/cpp/perfetto/producer_client.h

@@ -6,12 +6,15 @@
 #define SERVICES_TRACING_PUBLIC_CPP_PERFETTO_PRODUCER_CLIENT_H_
 
 #include <memory>
+#include <set>
 #include <string>
+#include <utility>
 #include <vector>
 
 #include "base/atomicops.h"
 #include "base/component_export.h"
 #include "base/macros.h"
+#include "base/memory/weak_ptr.h"
 #include "base/sequence_checker.h"
 #include "mojo/public/cpp/bindings/binding.h"
 #include "services/tracing/public/cpp/perfetto/task_runner.h"
@@ -41,6 +44,32 @@ class COMPONENT_EXPORT(TRACING_CPP) ProducerClient
     : public mojom::ProducerClient,
       public perfetto::TracingService::ProducerEndpoint {
  public:
+  class DataSourceBase {
+   public:
+    explicit DataSourceBase(const std::string& name);
+    virtual ~DataSourceBase();
+
+    void StartTracingWithID(uint64_t data_source_id,
+                            ProducerClient* producer_client,
+                            const mojom::DataSourceConfig& data_source_config);
+
+    virtual void StartTracing(
+        ProducerClient* producer_client,
+        const mojom::DataSourceConfig& data_source_config) = 0;
+    virtual void StopTracing(
+        base::OnceClosure stop_complete_callback = base::OnceClosure()) = 0;
+
+    // Flush the data source.
+    virtual void Flush(base::RepeatingClosure flush_complete_callback) = 0;
+
+    const std::string& name() const { return name_; }
+    uint64_t data_source_id() const { return data_source_id_; }
+
+   private:
+    uint64_t data_source_id_ = 0;
+    std::string name_;
+  };
+
   ProducerClient();
   ~ProducerClient() override;
 
@@ -58,6 +87,11 @@ class COMPONENT_EXPORT(TRACING_CPP) ProducerClient
                               mojom::ProducerHostRequest)>;
   void CreateMojoMessagepipes(MessagepipesReadyCallback);
 
+  // Add a new data source to the ProducerClient; the caller
+  // retains ownership and is responsible for making sure
+  // the data source outlives the ProducerClient.
+  void AddDataSource(DataSourceBase*);
+
   // mojom::ProducerClient implementation.
   // Called through Mojo by the ProducerHost on the service-side to control
   // tracing and toggle specific DataSources.
@@ -66,7 +100,9 @@ class COMPONENT_EXPORT(TRACING_CPP) ProducerClient
       uint64_t id,
       mojom::DataSourceConfigPtr data_source_config) override;
 
-  void TearDownDataSourceInstance(uint64_t id) override;
+  void TearDownDataSourceInstance(
+      uint64_t id,
+      TearDownDataSourceInstanceCallback callback) override;
   void Flush(uint64_t flush_request_id,
              const std::vector<uint64_t>& data_source_ids) override;
 
@@ -77,23 +113,25 @@ class COMPONENT_EXPORT(TRACING_CPP) ProducerClient
   void CommitData(const perfetto::CommitDataRequest& commit,
                   CommitDataCallback callback) override;
   // Used by the DataSource implementations to create TraceWriters
-  // for writing their protobufs
+  // for writing their protobufs, and respond to flushes.
   std::unique_ptr<perfetto::TraceWriter> CreateTraceWriter(
       perfetto::BufferID target_buffer) override;
+  void NotifyFlushComplete(perfetto::FlushRequestID) override;
   perfetto::SharedMemory* shared_memory() const override;
 
   // These ProducerEndpoint functions are only used on the service
   // side and should not be called on the clients.
   void RegisterDataSource(const perfetto::DataSourceDescriptor&) override;
   void UnregisterDataSource(const std::string& name) override;
-  size_t shared_buffer_page_size_kb() const override;
-  void NotifyFlushComplete(perfetto::FlushRequestID) override;
   void NotifyDataSourceStopped(perfetto::DataSourceInstanceID) override;
+  size_t shared_buffer_page_size_kb() const override;
 
   static void ResetTaskRunnerForTesting();
 
  private:
   void CommitDataOnSequence(mojom::CommitDataRequestPtr request);
+  void AddDataSourceOnSequence(DataSourceBase*);
+
   // The callback will be run on the |origin_task_runner|, meaning
   // the same sequence as CreateMojoMessagePipes() got called on.
   void CreateMojoMessagepipesOnSequence(
@@ -106,8 +144,15 @@ class COMPONENT_EXPORT(TRACING_CPP) ProducerClient
   std::unique_ptr<perfetto::SharedMemoryArbiter> shared_memory_arbiter_;
   mojom::ProducerHostPtr producer_host_;
   std::unique_ptr<MojoSharedMemory> shared_memory_;
+  std::set<DataSourceBase*> data_sources_;
+  // First value is the flush ID, the second is the number of
+  // replies we're still waiting for.
+  std::pair<uint64_t, size_t> pending_replies_for_latest_flush_;
+
   SEQUENCE_CHECKER(sequence_checker_);
 
+  // NOTE: Weak pointers must be invalidated before all other member variables.
+  base::WeakPtrFactory<ProducerClient> weak_ptr_factory_;
   DISALLOW_COPY_AND_ASSIGN(ProducerClient);
 };
 

services/tracing/public/cpp/perfetto/trace_event_data_source.cc

@@ -6,6 +6,8 @@
 
 #include <utility>
 
+#include "base/json/json_writer.h"
+#include "base/memory/ref_counted_memory.h"
 #include "base/no_destructor.h"
 #include "base/process/process_handle.h"
 #include "base/trace_event/trace_event.h"
@@ -21,6 +23,78 @@ using TraceConfig = base::trace_event::TraceConfig;
 
 namespace tracing {
 
+TraceEventMetadataSource::TraceEventMetadataSource()
+    : DataSourceBase(mojom::kMetaDataSourceName),
+      origin_task_runner_(base::SequencedTaskRunnerHandle::Get()) {}
+
+TraceEventMetadataSource::~TraceEventMetadataSource() = default;
+
+void TraceEventMetadataSource::AddGeneratorFunction(
+    MetadataGeneratorFunction generator) {
+  DCHECK(origin_task_runner_->RunsTasksInCurrentSequence());
+  base::AutoLock lock(lock_);
+  generator_functions_.push_back(generator);
+}
+
+void TraceEventMetadataSource::GenerateMetadata(
+    std::unique_ptr<perfetto::TraceWriter> trace_writer) {
+  DCHECK(origin_task_runner_->RunsTasksInCurrentSequence());
+
+  auto trace_packet = trace_writer->NewTracePacket();
+  protozero::MessageHandle<perfetto::protos::pbzero::ChromeEventBundle>
+      event_bundle(trace_packet->set_chrome_events());
+
+  base::AutoLock lock(lock_);
+  for (auto& generator : generator_functions_) {
+    std::unique_ptr<base::DictionaryValue> metadata_dict = generator.Run();
+    if (!metadata_dict) {
+      continue;
+    }
+
+    for (const auto& it : metadata_dict->DictItems()) {
+      auto* new_metadata = event_bundle->add_metadata();
+      new_metadata->set_name(it.first.c_str());
+
+      if (it.second.is_int()) {
+        new_metadata->set_int_value(it.second.GetInt());
+      } else if (it.second.is_bool()) {
+        new_metadata->set_bool_value(it.second.GetBool());
+      } else if (it.second.is_string()) {
+        new_metadata->set_string_value(it.second.GetString().c_str());
+      } else {
+        std::string json_value;
+        base::JSONWriter::Write(it.second, &json_value);
+        new_metadata->set_json_value(json_value.c_str());
+      }
+    }
+  }
+}
+
+void TraceEventMetadataSource::StartTracing(
+    ProducerClient* producer_client,
+    const mojom::DataSourceConfig& data_source_config) {
+  origin_task_runner_->PostTask(
+      FROM_HERE, base::BindOnce(&TraceEventMetadataSource::GenerateMetadata,
+                                base::Unretained(this),
+                                producer_client->CreateTraceWriter(
+                                    data_source_config.target_buffer)));
+}
+
+void TraceEventMetadataSource::StopTracing(
+    base::OnceClosure stop_complete_callback) {
+  // We bounce a task off the origin_task_runner_ that the generator
+  // callbacks are run from, to make sure that GenerateMetaData() has finished
+  // running.
+  origin_task_runner_->PostTaskAndReply(FROM_HERE, base::DoNothing(),
+                                        std::move(stop_complete_callback));
+}
+
+void TraceEventMetadataSource::Flush(
+    base::RepeatingClosure flush_complete_callback) {
+  origin_task_runner_->PostTaskAndReply(FROM_HERE, base::DoNothing(),
+                                        std::move(flush_complete_callback));
+}
+
 class TraceEventDataSource::ThreadLocalEventSink {
  public:
   explicit ThreadLocalEventSink(
@@ -169,6 +243,8 @@ class TraceEventDataSource::ThreadLocalEventSink {
     }
   }
 
+  void Flush() { trace_writer_->Flush(); }
+
  private:
   std::unique_ptr<perfetto::TraceWriter> trace_writer_;
 };
@@ -193,49 +269,86 @@ TraceEventDataSource* TraceEventDataSource::GetInstance() {
   return instance.get();
 }
 
-TraceEventDataSource::TraceEventDataSource() = default;
+TraceEventDataSource::TraceEventDataSource()
+    : DataSourceBase(mojom::kTraceEventDataSourceName) {}
 
 TraceEventDataSource::~TraceEventDataSource() = default;
 
 void TraceEventDataSource::StartTracing(
     ProducerClient* producer_client,
     const mojom::DataSourceConfig& data_source_config) {
+  {
     base::AutoLock lock(lock_);
 
     DCHECK(!producer_client_);
     producer_client_ = producer_client;
     target_buffer_ = data_source_config.target_buffer;
+  }
 
   TraceLog::GetInstance()->SetAddTraceEventOverride(
-      &TraceEventDataSource::OnAddTraceEvent);
+      &TraceEventDataSource::OnAddTraceEvent,
+      &TraceEventDataSource::FlushCurrentThread);
 
   TraceLog::GetInstance()->SetEnabled(
       TraceConfig(data_source_config.trace_config), TraceLog::RECORDING_MODE);
 }
 
 void TraceEventDataSource::StopTracing(
-    base::RepeatingClosure stop_complete_callback) {
-  DCHECK(producer_client_);
-
-  {
-    base::AutoLock lock(lock_);
+    base::OnceClosure stop_complete_callback) {
+  stop_complete_callback_ = std::move(stop_complete_callback);
 
-    producer_client_ = nullptr;
-    target_buffer_ = 0;
+  auto on_tracing_stopped_callback =
+      [](TraceEventDataSource* data_source,
+         const scoped_refptr<base::RefCountedString>&, bool has_more_events) {
+        if (has_more_events) {
+          return;
         }
 
-  TraceLog::GetInstance()->SetAddTraceEventOverride(nullptr);
+        TraceLog::GetInstance()->SetAddTraceEventOverride(nullptr, nullptr);
+
+        // TraceLog::CancelTracing will cause metadata events to be written;
+        // make sure we flush the TraceWriter for this thread (TraceLog will
+        // only call TraceEventDataSource::FlushCurrentThread for threads with
+        // a MessageLoop).
+        // TODO(oysteine): The perfetto service itself should be able to recover
+        // unreturned chunks so technically this can go away
+        // at some point, but seems needed for now.
+        FlushCurrentThread();
 
+        if (data_source->stop_complete_callback_) {
+          std::move(data_source->stop_complete_callback_).Run();
+        }
+      };
+
+  if (TraceLog::GetInstance()->IsEnabled()) {
     // We call CancelTracing because we don't want/need TraceLog to do any of
-  // its own JSON serialization on its own
+    // its own JSON serialization on its own.
     TraceLog::GetInstance()->CancelTracing(base::BindRepeating(
-      [](base::RepeatingClosure stop_complete_callback,
+        on_tracing_stopped_callback, base::Unretained(this)));
+  } else {
+    on_tracing_stopped_callback(this, scoped_refptr<base::RefCountedString>(),
+                                false);
+  }
+
+  base::AutoLock lock(lock_);
+  DCHECK(producer_client_);
+  producer_client_ = nullptr;
+  target_buffer_ = 0;
+}
+
+void TraceEventDataSource::Flush(
+    base::RepeatingClosure flush_complete_callback) {
+  DCHECK(TraceLog::GetInstance()->IsEnabled());
+  TraceLog::GetInstance()->Flush(base::BindRepeating(
+      [](base::RepeatingClosure flush_complete_callback,
          const scoped_refptr<base::RefCountedString>&, bool has_more_events) {
-        if (!has_more_events && stop_complete_callback) {
-          stop_complete_callback.Run();
+        if (has_more_events) {
+          return;
         }
+
+        flush_complete_callback.Run();
       },
-      std::move(stop_complete_callback)));
+      std::move(flush_complete_callback)));
 }
 
 TraceEventDataSource::ThreadLocalEventSink*
@@ -266,10 +379,11 @@ void TraceEventDataSource::OnAddTraceEvent(const TraceEvent& trace_event) {
 }
 
 // static
-void TraceEventDataSource::ResetCurrentThreadForTesting() {
-  ThreadLocalEventSink* thread_local_event_sink =
+void TraceEventDataSource::FlushCurrentThread() {
+  auto* thread_local_event_sink =
       static_cast<ThreadLocalEventSink*>(ThreadLocalEventSinkSlot()->Get());
   if (thread_local_event_sink) {
+    thread_local_event_sink->Flush();
     delete thread_local_event_sink;
     ThreadLocalEventSinkSlot()->Set(nullptr);
   }

services/tracing/public/cpp/perfetto/trace_event_data_source.h

@@ -7,55 +7,89 @@
 
 #include <memory>
 #include <string>
+#include <vector>
 
 #include "base/component_export.h"
 #include "base/macros.h"
 #include "base/threading/thread_local.h"
 #include "services/tracing/public/cpp/perfetto/producer_client.h"
 
+namespace perfetto {
+class TraceWriter;
+}
+
 namespace tracing {
 
-class ProducerClient;
+// This class is a data source that clients can use to provide
+// global metadata in dictionary form, by registering callbacks.
+class COMPONENT_EXPORT(TRACING_CPP) TraceEventMetadataSource
+    : public ProducerClient::DataSourceBase {
+ public:
+  TraceEventMetadataSource();
+  ~TraceEventMetadataSource() override;
+
+  using MetadataGeneratorFunction =
+      base::RepeatingCallback<std::unique_ptr<base::DictionaryValue>()>;
+  // Any callbacks passed here will be called when tracing starts.
+  void AddGeneratorFunction(MetadataGeneratorFunction generator);
+
+  // ProducerClient::DataSourceBase implementation, called by
+  // ProducerClent.
+  void StartTracing(ProducerClient* producer_client,
+                    const mojom::DataSourceConfig& data_source_config) override;
+  void StopTracing(base::OnceClosure stop_complete_callback) override;
+  void Flush(base::RepeatingClosure flush_complete_callback) override;
+
+ private:
+  void GenerateMetadata(std::unique_ptr<perfetto::TraceWriter> trace_writer);
+
+  std::vector<MetadataGeneratorFunction> generator_functions_;
+  scoped_refptr<base::SequencedTaskRunner> origin_task_runner_;
+  base::Lock lock_;
+
+  DISALLOW_COPY_AND_ASSIGN(TraceEventMetadataSource);
+};
 
 // This class acts as a bridge between the TraceLog and
 // the Perfetto ProducerClient. It converts incoming
 // trace events to ChromeTraceEvent protos and writes
 // them into the Perfetto shared memory.
-class COMPONENT_EXPORT(TRACING_CPP) TraceEventDataSource {
+class COMPONENT_EXPORT(TRACING_CPP) TraceEventDataSource
+    : public ProducerClient::DataSourceBase {
  public:
   class ThreadLocalEventSink;
 
   static TraceEventDataSource* GetInstance();
 
-  // Deletes the TraceWriter for the current thread, if any.
-  static void ResetCurrentThreadForTesting();
+  // Flushes and deletes the TraceWriter for the current thread, if any.
+  static void FlushCurrentThread();
 
-  // The ProducerClient is responsible for calling RequestStop
+  // The ProducerClient is responsible for calling StopTracing
   // which will clear the stored pointer to it, before it
   // gets destroyed. ProducerClient::CreateTraceWriter can be
   // called by the TraceEventDataSource on any thread.
   void StartTracing(ProducerClient* producer_client,
-                    const mojom::DataSourceConfig& data_source_config);
+                    const mojom::DataSourceConfig& data_source_config) override;
 
   // Called from the ProducerClient.
-  void StopTracing(
-      base::RepeatingClosure stop_complete_callback = base::RepeatingClosure());
+  void StopTracing(base::OnceClosure stop_complete_callback) override;
+  void Flush(base::RepeatingClosure flush_complete_callback) override;
 
  private:
   friend class base::NoDestructor<TraceEventDataSource>;
 
   TraceEventDataSource();
-  ~TraceEventDataSource();
+  ~TraceEventDataSource() override;
 
   ThreadLocalEventSink* CreateThreadLocalEventSink();
 
-  // Callback from TraceLog on any added trace events, can be called from
-  // any thread.
+  // Callback from TraceLog, can be called from any thread.
   static void OnAddTraceEvent(const base::trace_event::TraceEvent& trace_event);
 
   base::Lock lock_;
   uint32_t target_buffer_ = 0;
   ProducerClient* producer_client_ = nullptr;
+  base::OnceClosure stop_complete_callback_;
 
   DISALLOW_COPY_AND_ASSIGN(TraceEventDataSource);
 };

services/tracing/public/cpp/perfetto/trace_event_data_source_unittest.cc

@@ -10,6 +10,7 @@
 #include "base/bind.h"
 #include "base/callback.h"
 #include "base/debug/leak_annotations.h"
+#include "base/json/json_reader.h"
 #include "base/message_loop/message_loop.h"
 #include "base/run_loop.h"
 #include "base/test/scoped_task_environment.h"
@@ -30,10 +31,12 @@ const char kCategoryGroup[] = "foo";
 class MockProducerClient : public ProducerClient {
  public:
   explicit MockProducerClient(
-      scoped_refptr<base::SequencedTaskRunner> main_thread_task_runner)
+      scoped_refptr<base::SequencedTaskRunner> main_thread_task_runner,
+      const char* wanted_event_category)
       : delegate_(perfetto::base::kPageSize),
         stream_(&delegate_),
-        main_thread_task_runner_(std::move(main_thread_task_runner)) {
+        main_thread_task_runner_(std::move(main_thread_task_runner)),
+        wanted_event_category_(wanted_event_category) {
     trace_packet_.Reset(&stream_);
   }
 
@@ -55,8 +58,11 @@ class MockProducerClient : public ProducerClient {
       if (proto->has_chrome_events() &&
           proto->chrome_events().trace_events().size() > 0 &&
           proto->chrome_events().trace_events()[0].category_group_name() ==
-              kCategoryGroup) {
+              wanted_event_category_) {
         finalized_packets_.push_back(std::move(proto));
+      } else if (proto->has_chrome_events() &&
+                 proto->chrome_events().metadata().size() > 0) {
+        metadata_packets_.push_back(std::move(proto));
       }
     }
 
@@ -84,13 +90,24 @@ class MockProducerClient : public ProducerClient {
     return event_bundle.trace_events();
   }
 
+  const google::protobuf::RepeatedPtrField<perfetto::protos::ChromeMetadata>
+  GetChromeMetadata(size_t packet_index = 0) {
+    FlushPacketIfPossible();
+    EXPECT_GT(metadata_packets_.size(), packet_index);
+
+    auto event_bundle = metadata_packets_[packet_index]->chrome_events();
+    return event_bundle.metadata();
+  }
+
  private:
   std::vector<std::unique_ptr<perfetto::protos::TracePacket>>
       finalized_packets_;
+  std::vector<std::unique_ptr<perfetto::protos::TracePacket>> metadata_packets_;
   perfetto::protos::pbzero::TracePacket trace_packet_;
   protozero::ScatteredStreamWriterNullDelegate delegate_;
   protozero::ScatteredStreamWriter stream_;
   scoped_refptr<base::SequencedTaskRunner> main_thread_task_runner_;
+  const char* wanted_event_category_;
 };
 
 // For sequences/threads other than our own, we just want to ignore
@@ -157,27 +174,34 @@ class TraceEventDataSourceTest : public testing::Test {
  public:
   void SetUp() override {
     ProducerClient::ResetTaskRunnerForTesting();
-    producer_client_ = std::make_unique<MockProducerClient>(
-        scoped_task_environment_.GetMainThreadTaskRunner());
   }
 
   void TearDown() override {
+    if (base::trace_event::TraceLog::GetInstance()->IsEnabled()) {
       base::RunLoop wait_for_tracelog_flush;
 
       TraceEventDataSource::GetInstance()->StopTracing(base::BindRepeating(
-        [](const base::RepeatingClosure& quit_closure) { quit_closure.Run(); },
+          [](const base::RepeatingClosure& quit_closure) {
+            quit_closure.Run();
+          },
           wait_for_tracelog_flush.QuitClosure()));
 
       wait_for_tracelog_flush.Run();
+    }
 
     // As MockTraceWriter keeps a pointer to our MockProducerClient,
     // we need to make sure to clean it up from TLS. The other sequences
     // get DummyTraceWriters that we don't care about.
-    TraceEventDataSource::GetInstance()->ResetCurrentThreadForTesting();
+    TraceEventDataSource::GetInstance()->FlushCurrentThread();
     producer_client_.reset();
   }
 
-  void CreateTraceEventDataSource() {
+  void CreateTraceEventDataSource(
+      const char* wanted_event_category = kCategoryGroup) {
+    producer_client_ = std::make_unique<MockProducerClient>(
+        scoped_task_environment_.GetMainThreadTaskRunner(),
+        wanted_event_category);
+
     auto data_source_config = mojom::DataSourceConfig::New();
     TraceEventDataSource::GetInstance()->StartTracing(producer_client(),
                                                       *data_source_config);
@@ -190,6 +214,105 @@ class TraceEventDataSourceTest : public testing::Test {
   base::test::ScopedTaskEnvironment scoped_task_environment_;
 };
 
+void HasMetadataValue(const perfetto::protos::ChromeMetadata& entry,
+                      const char* value) {
+  EXPECT_TRUE(entry.has_string_value());
+  EXPECT_EQ(entry.string_value(), value);
+}
+
+void HasMetadataValue(const perfetto::protos::ChromeMetadata& entry,
+                      int value) {
+  EXPECT_TRUE(entry.has_int_value());
+  EXPECT_EQ(entry.int_value(), value);
+}
+
+void HasMetadataValue(const perfetto::protos::ChromeMetadata& entry,
+                      bool value) {
+  EXPECT_TRUE(entry.has_bool_value());
+  EXPECT_EQ(entry.bool_value(), value);
+}
+
+void HasMetadataValue(const perfetto::protos::ChromeMetadata& entry,
+                      const base::DictionaryValue& value) {
+  EXPECT_TRUE(entry.has_json_value());
+
+  std::unique_ptr<base::Value> child_dict =
+      base::JSONReader::Read(entry.json_value());
+  EXPECT_EQ(*child_dict, value);
+}
+
+template <typename T>
+void MetadataHasNamedValue(const google::protobuf::RepeatedPtrField<
+                               perfetto::protos::ChromeMetadata>& metadata,
+                           const char* name,
+                           const T& value) {
+  for (int i = 0; i < metadata.size(); i++) {
+    auto& entry = metadata[i];
+    if (entry.name() == name) {
+      HasMetadataValue(entry, value);
+      return;
+    }
+  }
+
+  NOTREACHED();
+}
+
+TEST_F(TraceEventDataSourceTest, MetadataSourceBasicTypes) {
+  auto metadata_source = std::make_unique<TraceEventMetadataSource>();
+  metadata_source->AddGeneratorFunction(base::BindRepeating([]() {
+    auto metadata = std::make_unique<base::DictionaryValue>();
+    metadata->SetInteger("foo_int", 42);
+    metadata->SetString("foo_str", "bar");
+    metadata->SetBoolean("foo_bool", true);
+
+    auto child_dict = std::make_unique<base::DictionaryValue>();
+    child_dict->SetString("child_str", "child_val");
+    metadata->Set("child_dict", std::move(child_dict));
+    return metadata;
+  }));
+
+  CreateTraceEventDataSource();
+
+  auto data_source_config = mojom::DataSourceConfig::New();
+  metadata_source->StartTracing(producer_client(), *data_source_config);
+
+  base::RunLoop wait_for_flush;
+  metadata_source->Flush(wait_for_flush.QuitClosure());
+  wait_for_flush.Run();
+
+  auto metadata = producer_client()->GetChromeMetadata();
+  EXPECT_EQ(4, metadata.size());
+  MetadataHasNamedValue(metadata, "foo_int", 42);
+  MetadataHasNamedValue(metadata, "foo_str", "bar");
+  MetadataHasNamedValue(metadata, "foo_bool", true);
+
+  auto child_dict = std::make_unique<base::DictionaryValue>();
+  child_dict->SetString("child_str", "child_val");
+  MetadataHasNamedValue(metadata, "child_dict", *child_dict);
+}
+
+TEST_F(TraceEventDataSourceTest, TraceLogMetadataEvents) {
+  CreateTraceEventDataSource("__metadata");
+
+  base::RunLoop wait_for_flush;
+  TraceEventDataSource::GetInstance()->StopTracing(
+      wait_for_flush.QuitClosure());
+  wait_for_flush.Run();
+
+  bool has_process_uptime_event = false;
+  for (size_t i = 0; i < producer_client()->GetFinalizedPacketCount(); ++i) {
+    auto trace_events = producer_client()->GetChromeTraceEvents(i);
+    for (auto& event : trace_events) {
+      if (event.name() == "process_uptime_seconds") {
+        has_process_uptime_event = true;
+        break;
+      }
+    }
+  }
+
+  EXPECT_TRUE(has_process_uptime_event);
+}
+
 TEST_F(TraceEventDataSourceTest, BasicTraceEvent) {
   CreateTraceEventDataSource();
 

services/tracing/public/cpp/trace_event_agent.cc

@@ -26,6 +26,10 @@
     defined(OS_WIN)
 #define PERFETTO_AVAILABLE
 #include "services/tracing/public/cpp/perfetto/producer_client.h"
+#include "services/tracing/public/cpp/perfetto/trace_event_data_source.h"
+#include "third_party/perfetto/include/perfetto/tracing/core/trace_writer.h"
+#include "third_party/perfetto/protos/perfetto/trace/chrome/chrome_trace_event.pbzero.h"
+#include "third_party/perfetto/protos/perfetto/trace/trace_packet.pbzero.h"
 #endif
 
 namespace {
@@ -63,6 +67,21 @@ class PerfettoTraceEventAgent : public TraceEventAgent {
               std::move(producer_client_pipe), std::move(producer_host_pipe));
         },
         std::move(perfetto_service)));
+
+    GetProducerClient()->AddDataSource(TraceEventDataSource::GetInstance());
+  }
+
+  void AddMetadataGeneratorFunction(
+      MetadataGeneratorFunction generator) override {
+    // Instantiate and register the metadata data source on the first
+    // call.
+    static TraceEventMetadataSource* metadata_source = []() {
+      static base::NoDestructor<TraceEventMetadataSource> instance;
+      GetProducerClient()->AddDataSource(instance.get());
+      return instance.get();
+    }();
+
+    metadata_source->AddGeneratorFunction(generator);
   }
 };
 #endif

services/tracing/public/cpp/trace_event_agent.h

@@ -48,7 +48,7 @@ class COMPONENT_EXPORT(TRACING_CPP) TraceEventAgent : public BaseAgent {
   void GetCategories(GetCategoriesCallback callback) override;
 
   virtual void AddMetadataGeneratorFunction(
-      MetadataGeneratorFunction generator) {}
+      MetadataGeneratorFunction generator) = 0;
 
  private:
   DISALLOW_COPY_AND_ASSIGN(TraceEventAgent);

services/tracing/public/mojom/perfetto_service.mojom

@@ -5,6 +5,7 @@
 module tracing.mojom;
 
 const string kTraceEventDataSourceName = "org.chromium.trace_event";
+const string kMetaDataSourceName = "org.chromium.trace_metadata";
 
 // Brief description of the flow: There's a per-process ProducerClient
 // which connects to the central PerfettoService and establishes a two-way
@@ -57,6 +58,7 @@ struct ChunksToPatch {
 struct CommitDataRequest {
   array<ChunksToMove> chunks_to_move;
   array<ChunksToPatch> chunks_to_patch;
+  uint64 flush_request_id;
 };
 
 struct DataSourceConfig {
@@ -65,25 +67,39 @@ struct DataSourceConfig {
   uint32 target_buffer;
 };
 
+struct DataSourceRegistration {
+  string name;
+  bool will_notify_on_stop;
+};
+
 interface ProducerHost {
   // Called by a ProducerClient to asks the service to:
-  // 1) Move data from the shared memory buffer into the final tracing buffer owned by the service
-  //    (through the |chunks_to_move|).
-  // 2) Patch data (i.e. apply diff) that has been previously copied into the tracing buffer
-  //    (if it's not been overwritten).
-  // The service is robust in terms of tolerating malformed or malicious requests.
+  // 1) Move data from the shared memory buffer into the final tracing buffer
+  //    owned by the service (through the |chunks_to_move|).
+  // 2) Patch data (i.e. apply diff) that has been previously copied into the
+  //    tracing buffer (if it's not been overwritten).
+  // The service is robust in terms of tolerating malformed or malicious
+  // requests.
   CommitData(CommitDataRequest data_request);
+
+  // Called by a ProducerClient to let the Host know it can provide a
+  // specific datasource.
+  RegisterDataSource(DataSourceRegistration registration_info);
+
+  // Called to let the Service know that a flush is complete.
+  NotifyFlushComplete(uint64 flush_request_id);
 };
 
 interface ProducerClient {
     OnTracingStart(handle<shared_buffer> shared_memory);
 
     // TODO(oysteine): Make a TypeTrait for sending the full DataSourceConfig.
-    // Called by Perfetto (via ProducerHost) to request a data source to start logging.
+    // Called by Perfetto (via ProducerHost) to request a data source to start
+    // logging.
     CreateDataSourceInstance(uint64 id, DataSourceConfig data_source_config);
-    // Requesting a data source to stop logging again, with the id previously sent
-    // in the CreateDataSourceInstance call.
-    TearDownDataSourceInstance(uint64 id);
+    // Requesting a data source to stop logging again, with the id previously
+    // sent in the CreateDataSourceInstance call.
+    TearDownDataSourceInstance(uint64 id) => ();
     Flush(uint64 flush_request_id, array<uint64> data_source_ids);
 };