src/traced/probes/probes_producer.cc

/*
 * Copyright (C) 2018 The Android Open Source Project
 *
 * 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/traced/probes/probes_producer.h"

#include <stdio.h>
#include <sys/stat.h>

#include <algorithm>
#include <queue>
#include <string>

#include "perfetto/base/logging.h"
#include "perfetto/base/weak_ptr.h"
#include "perfetto/traced/traced.h"
#include "perfetto/tracing/core/data_source_config.h"
#include "perfetto/tracing/core/data_source_descriptor.h"
#include "perfetto/tracing/core/ftrace_config.h"
#include "perfetto/tracing/core/trace_config.h"
#include "perfetto/tracing/core/trace_packet.h"
#include "src/traced/probes/filesystem/inode_file_data_source.h"

#include "perfetto/trace/filesystem/inode_file_map.pbzero.h"
#include "perfetto/trace/ftrace/ftrace_event_bundle.pbzero.h"
#include "perfetto/trace/trace_packet.pbzero.h"

namespace perfetto {
namespace {

constexpr uint32_t kInitialConnectionBackoffMs = 100;
constexpr uint32_t kMaxConnectionBackoffMs = 30 * 1000;
constexpr char kFtraceSourceName[] = "linux.ftrace";
constexpr char kProcessStatsSourceName[] = "linux.process_stats";
constexpr char kInodeMapSourceName[] = "linux.inode_file_map";

}  // namespace.

// State transition diagram:
//                    +----------------------------+
//                    v                            +
// NotStarted -> NotConnected -> Connecting -> Connected
//                    ^              +
//                    +--------------+
//

ProbesProducer::ProbesProducer() {}
ProbesProducer::~ProbesProducer() = default;

void ProbesProducer::OnConnect() {
  PERFETTO_DCHECK(state_ == kConnecting);
  state_ = kConnected;
  ResetConnectionBackoff();
  PERFETTO_LOG("Connected to the service");

  DataSourceDescriptor ftrace_descriptor;
  ftrace_descriptor.set_name(kFtraceSourceName);
  endpoint_->RegisterDataSource(ftrace_descriptor);

  DataSourceDescriptor process_stats_descriptor;
  process_stats_descriptor.set_name(kProcessStatsSourceName);
  endpoint_->RegisterDataSource(process_stats_descriptor);

  DataSourceDescriptor inode_map_descriptor;
  inode_map_descriptor.set_name(kInodeMapSourceName);
  endpoint_->RegisterDataSource(inode_map_descriptor);
}

void ProbesProducer::OnDisconnect() {
  PERFETTO_DCHECK(state_ == kConnected || state_ == kConnecting);
  PERFETTO_LOG("Disconnected from tracing service");
  if (state_ == kConnected)
    return task_runner_->PostTask([this] { this->Restart(); });

  state_ = kNotConnected;
  IncreaseConnectionBackoff();
  task_runner_->PostDelayedTask([this] { this->Connect(); },
                                connection_backoff_ms_);
}

void ProbesProducer::Restart() {
  // We lost the connection with the tracing service. At this point we need
  // to reset all the data sources. Trying to handle that manually is going to
  // be error prone. What we do here is simply desroying the instance and
  // recreating it again.
  // TODO(hjd): Add e2e test for this.

  base::TaskRunner* task_runner = task_runner_;
  const char* socket_name = socket_name_;

  // Invoke destructor and then the constructor again.
  this->~ProbesProducer();
  new (this) ProbesProducer();

  ConnectWithRetries(socket_name, task_runner);
}

void ProbesProducer::CreateDataSourceInstance(DataSourceInstanceID instance_id,
                                              const DataSourceConfig& config) {
  // TODO(hjd): This a hack since we don't actually know the session id. For
  // now we'll assume anything wit hthe same target buffer is in the same
  // session.
  TracingSessionID session_id = config.target_buffer();

  if (config.name() == kFtraceSourceName) {
    if (!CreateFtraceDataSourceInstance(session_id, instance_id, config))
      failed_sources_.insert(instance_id);
  } else if (config.name() == kInodeMapSourceName) {
    CreateInodeFileDataSourceInstance(session_id, instance_id, config);
  } else if (config.name() == kProcessStatsSourceName) {
    CreateProcessStatsDataSourceInstance(session_id, instance_id, config);
  } else {
    PERFETTO_ELOG("Data source name: %s not recognised.",
                  config.name().c_str());
    return;
  }

  std::map<TracingSessionID, InodeFileDataSource*> file_sources;
  std::map<TracingSessionID, ProcessStatsDataSource*> ps_sources;
  for (const auto& pair : file_map_sources_)
    file_sources[pair.second->session_id()] = pair.second.get();
  for (const auto& pair : process_stats_sources_)
    ps_sources[pair.second->session_id()] = pair.second.get();

  for (const auto& id_to_source : delegates_) {
    const std::unique_ptr<SinkDelegate>& source = id_to_source.second;
    if (session_id != source->session_id())
      continue;
    if (!source->ps_source() && ps_sources.count(session_id))
      source->set_ps_source(ps_sources[session_id]->GetWeakPtr());
    if (!source->file_source() && file_sources.count(session_id))
      source->set_file_source(file_sources[session_id]->GetWeakPtr());
  }
}

void ProbesProducer::AddWatchdogsTimer(DataSourceInstanceID id,
                                       const DataSourceConfig& config) {
  if (config.trace_duration_ms() != 0)
    watchdogs_.emplace(id, base::Watchdog::GetInstance()->CreateFatalTimer(
                               5000 + 2 * config.trace_duration_ms()));
}

bool ProbesProducer::CreateFtraceDataSourceInstance(
    TracingSessionID session_id,
    DataSourceInstanceID id,
    const DataSourceConfig& config) {
  // Don't retry if FtraceController::Create() failed once.
  // This can legitimately happen on user builds where we cannot access the
  // debug paths, e.g., because of SELinux rules.
  if (ftrace_creation_failed_)
    return false;

  // Lazily create on the first instance.
  if (!ftrace_) {
    ftrace_ = FtraceController::Create(task_runner_);

    if (!ftrace_) {
      PERFETTO_ELOG("Failed to create FtraceController");
      ftrace_creation_failed_ = true;
      return false;
    }

    ftrace_->DisableAllEvents();
    ftrace_->ClearTrace();
  }

  PERFETTO_LOG("Ftrace start (id=%" PRIu64 ", target_buf=%" PRIu32 ")", id,
               config.target_buffer());

  FtraceConfig proto_config = config.ftrace_config();

  // TODO(hjd): Static cast is bad, target_buffer() should return a BufferID.
  auto trace_writer = endpoint_->CreateTraceWriter(
      static_cast<BufferID>(config.target_buffer()));
  auto delegate = std::unique_ptr<SinkDelegate>(
      new SinkDelegate(session_id, task_runner_, std::move(trace_writer)));
  auto sink = ftrace_->CreateSink(std::move(proto_config), delegate.get());
  if (!sink) {
    PERFETTO_ELOG("Failed to start tracing (maybe someone else is using it?)");
    return false;
  }
  delegate->set_sink(std::move(sink));
  delegates_.emplace(id, std::move(delegate));
  AddWatchdogsTimer(id, config);
  return true;
}

void ProbesProducer::CreateInodeFileDataSourceInstance(
    TracingSessionID session_id,
    DataSourceInstanceID id,
    DataSourceConfig source_config) {
  PERFETTO_LOG("Inode file map start (id=%" PRIu64 ", target_buf=%" PRIu32 ")",
               id, source_config.target_buffer());
  auto trace_writer = endpoint_->CreateTraceWriter(
      static_cast<BufferID>(source_config.target_buffer()));
  if (system_inodes_.empty())
    CreateStaticDeviceToInodeMap("/system", &system_inodes_);
  auto file_map_source =
      std::unique_ptr<InodeFileDataSource>(new InodeFileDataSource(
          std::move(source_config), task_runner_, session_id, &system_inodes_,
          &cache_, std::move(trace_writer)));
  file_map_sources_.emplace(id, std::move(file_map_source));
  AddWatchdogsTimer(id, source_config);
}

void ProbesProducer::CreateProcessStatsDataSourceInstance(
    TracingSessionID session_id,
    DataSourceInstanceID id,
    const DataSourceConfig& config) {
  PERFETTO_DCHECK(process_stats_sources_.count(id) == 0);
  auto trace_writer = endpoint_->CreateTraceWriter(
      static_cast<BufferID>(config.target_buffer()));
  auto source = std::unique_ptr<ProcessStatsDataSource>(
      new ProcessStatsDataSource(session_id, std::move(trace_writer), config));
  auto it_and_inserted = process_stats_sources_.emplace(id, std::move(source));
  PERFETTO_DCHECK(it_and_inserted.second);
  const auto& quirks =
      it_and_inserted.first->second->config().process_stats_config().quirks();
  if (std::find(quirks.begin(), quirks.end(),
                ProcessStatsConfig::DISABLE_INITIAL_DUMP) != quirks.end()) {
    PERFETTO_DLOG("Initial process tree dump is disabled.");
    return;
  }
  it_and_inserted.first->second->WriteAllProcesses();
}

void ProbesProducer::TearDownDataSourceInstance(DataSourceInstanceID id) {
  PERFETTO_LOG("Producer stop (id=%" PRIu64 ")", id);
  // |id| could be the id of any of the datasources we handle:
  PERFETTO_DCHECK((failed_sources_.count(id) + delegates_.count(id) +
                   process_stats_sources_.count(id) +
                   file_map_sources_.count(id)) == 1);
  failed_sources_.erase(id);
  delegates_.erase(id);
  process_stats_sources_.erase(id);
  file_map_sources_.erase(id);
  watchdogs_.erase(id);
}

void ProbesProducer::OnTracingSetup() {}

void ProbesProducer::Flush(FlushRequestID flush_request_id,
                           const DataSourceInstanceID* data_source_ids,
                           size_t num_data_sources) {
  for (size_t i = 0; i < num_data_sources; i++) {
    DataSourceInstanceID ds_id = data_source_ids[i];
    {
      auto it = process_stats_sources_.find(ds_id);
      if (it != process_stats_sources_.end())
        it->second->Flush();
    }
    {
      auto it = file_map_sources_.find(ds_id);
      if (it != file_map_sources_.end())
        it->second->Flush();
    }
    {
      auto it = delegates_.find(ds_id);
      if (it != delegates_.end())
        it->second->Flush();
    }
  }
  endpoint_->NotifyFlushComplete(flush_request_id);
}

void ProbesProducer::ConnectWithRetries(const char* socket_name,
                                        base::TaskRunner* task_runner) {
  PERFETTO_DCHECK(state_ == kNotStarted);
  state_ = kNotConnected;

  ResetConnectionBackoff();
  socket_name_ = socket_name;
  task_runner_ = task_runner;
  Connect();
}

void ProbesProducer::Connect() {
  PERFETTO_DCHECK(state_ == kNotConnected);
  state_ = kConnecting;
  endpoint_ = ProducerIPCClient::Connect(
      socket_name_, this, "perfetto.traced_probes", task_runner_);
}

void ProbesProducer::IncreaseConnectionBackoff() {
  connection_backoff_ms_ *= 2;
  if (connection_backoff_ms_ > kMaxConnectionBackoffMs)
    connection_backoff_ms_ = kMaxConnectionBackoffMs;
}

void ProbesProducer::ResetConnectionBackoff() {
  connection_backoff_ms_ = kInitialConnectionBackoffMs;
}

ProbesProducer::SinkDelegate::SinkDelegate(TracingSessionID id,
                                           base::TaskRunner* task_runner,
                                           std::unique_ptr<TraceWriter> writer)
    : session_id_(id),
      task_runner_(task_runner),
      writer_(std::move(writer)),
      weak_factory_(this) {}

ProbesProducer::SinkDelegate::~SinkDelegate() = default;

void ProbesProducer::SinkDelegate::Flush() {
  // TODO(primiano): this still doesn't flush data from the kernel ftrace
  // buffers (see b/73886018). We should do that and delay the
  // NotifyFlushComplete() until the ftrace data has been drained from the
  // kernel ftrace buffer and written in the SMB.
  if (writer_ && (!trace_packet_ || trace_packet_->is_finalized()))
    writer_->Flush();
}

ProbesProducer::FtraceBundleHandle
ProbesProducer::SinkDelegate::GetBundleForCpu(size_t) {
  trace_packet_ = writer_->NewTracePacket();
  return FtraceBundleHandle(trace_packet_->set_ftrace_events());
}

void ProbesProducer::SinkDelegate::OnBundleComplete(
    size_t,
    FtraceBundleHandle,
    const FtraceMetadata& metadata) {
  trace_packet_->Finalize();

  if (file_source_ && !metadata.inode_and_device.empty()) {
    auto inodes = metadata.inode_and_device;
    auto weak_file_source = file_source_;
    task_runner_->PostTask([weak_file_source, inodes] {
      if (weak_file_source)
        weak_file_source->OnInodes(inodes);
    });
  }
  if (ps_source_ && !metadata.pids.empty()) {
    const auto& quirks = ps_source_->config().process_stats_config().quirks();
    if (std::find(quirks.begin(), quirks.end(),
                  ProcessStatsConfig::DISABLE_ON_DEMAND) != quirks.end()) {
      return;
    }
    const auto& pids = metadata.pids;
    auto weak_ps_source = ps_source_;
    task_runner_->PostTask([weak_ps_source, pids] {
      if (weak_ps_source)
        weak_ps_source->OnPids(pids);
    });
  }
}

}  // namespace perfetto