Method Index

src/brpc/span.cpp:692–805

}

leveldb::Status SpanDB::Index(std::shared_ptr<const Span> span, std::string* value_buf) {
    leveldb::WriteOptions options;
    options.sync = false;

    leveldb::Status st;

    // NOTE: Writing into time_db before id_db so that if the second write
    // fails, the entry in time_db will be finally removed when it's out
    // of time window.

    const int64_t start_time = span->GetStartRealTimeUs();
    const int64_t latency_us = span->GetEndRealTimeUs() - start_time;
    // if latency_us < FLAGS_rpcz_save_span_min_latency_us, don't save this span
    if (latency_us < FLAGS_rpcz_save_span_min_latency_us) {
        return leveldb::Status::OK();
    }
    BriefSpan brief;
    brief.set_trace_id(span->trace_id());
    brief.set_span_id(span->span_id());
    brief.set_log_id(span->log_id());
    brief.set_type(span->type());
    brief.set_error_code(span->error_code());
    brief.set_request_size(span->request_size());
    brief.set_response_size(span->response_size());
    brief.set_start_real_us(start_time);
    brief.set_latency_us(latency_us);
    brief.set_full_method_name(span->full_method_name());
    if (!brief.SerializeToString(value_buf)) {
        return leveldb::Status::InvalidArgument(
            leveldb::Slice("Fail to serialize BriefSpan"));
    }
    // We need to make the time monotonic otherwise if older entries are
    // overwritten by newer ones, entries in id_db associated with the older
    // entries are not evicted. Surely we can call DB::Get() before Put(), but
    // that would be too slow due to the storage model of leveldb. One feasible
    // method is to maintain recent window of keys to time_db, when there's a
    // conflict before Put(), try key+1us until an unused time is found. The
    // window could be 5~10s. However this method needs a std::map(slow) or
    // hashmap+queue(more memory: remember that we're just a framework), and
    // this method can't guarantee no duplication when real time goes back
    // significantly.
    // Since the time to this method is ALMOST in ascending order, we use a
    // very simple strategy: if the time is not greater than last-time, set
    // it to be last-time + 1us. This works when time goes back because the
    // real time is at least 1000000 / FLAGS_rpcz_max_span_per_second times faster
    // and it will finally catch up with our time key. (provided the flag
    // is less than 1000000).
    int64_t time_key = start_time;
    if (time_key <= g_last_time_key) {
        time_key = g_last_time_key + 1;
    }
    g_last_time_key = time_key;
    uint32_t time_data[2];
    ToBigEndian(time_key, time_data);
    st = time_db->Put(options,
                      leveldb::Slice((char*)time_data, sizeof(time_data)),
                      leveldb::Slice(value_buf->data(), value_buf->size()));
    if (!st.ok()) {