Use traceparent header to get tracing.

This commit is contained in:
Syfaro 2020-02-05 20:31:06 -06:00
parent af213d4365
commit 61f87e5526
3 changed files with 41 additions and 5 deletions

View File

@ -12,6 +12,7 @@ pub fn search(db: Pool) -> impl Filter<Extract = impl Reply, Error = Rejection>
pub fn search_file(db: Pool) -> impl Filter<Extract = impl Reply, Error = Rejection> + Clone { pub fn search_file(db: Pool) -> impl Filter<Extract = impl Reply, Error = Rejection> + Clone {
warp::path("file") warp::path("file")
.and(with_telem())
.and(warp::get()) .and(warp::get())
.and(warp::query::<FileSearchOpts>()) .and(warp::query::<FileSearchOpts>())
.and(with_pool(db)) .and(with_pool(db))
@ -21,6 +22,7 @@ pub fn search_file(db: Pool) -> impl Filter<Extract = impl Reply, Error = Reject
pub fn search_image(db: Pool) -> impl Filter<Extract = impl Reply, Error = Rejection> + Clone { pub fn search_image(db: Pool) -> impl Filter<Extract = impl Reply, Error = Rejection> + Clone {
warp::path("image") warp::path("image")
.and(with_telem())
.and(warp::post()) .and(warp::post())
.and(warp::multipart::form().max_length(1024 * 1024 * 10)) .and(warp::multipart::form().max_length(1024 * 1024 * 10))
.and(warp::query::<ImageSearchOpts>()) .and(warp::query::<ImageSearchOpts>())
@ -31,6 +33,7 @@ pub fn search_image(db: Pool) -> impl Filter<Extract = impl Reply, Error = Rejec
pub fn search_hashes(db: Pool) -> impl Filter<Extract = impl Reply, Error = Rejection> + Clone { pub fn search_hashes(db: Pool) -> impl Filter<Extract = impl Reply, Error = Rejection> + Clone {
warp::path("hashes") warp::path("hashes")
.and(with_telem())
.and(warp::get()) .and(warp::get())
.and(warp::query::<HashSearchOpts>()) .and(warp::query::<HashSearchOpts>())
.and(with_pool(db)) .and(with_pool(db))
@ -42,6 +45,7 @@ pub fn stream_search_image(
db: Pool, db: Pool,
) -> impl Filter<Extract = impl Reply, Error = Rejection> + Clone { ) -> impl Filter<Extract = impl Reply, Error = Rejection> + Clone {
warp::path("stream") warp::path("stream")
.and(with_telem())
.and(warp::post()) .and(warp::post())
.and(warp::multipart::form().max_length(1024 * 1024 * 10)) .and(warp::multipart::form().max_length(1024 * 1024 * 10))
.and(with_pool(db)) .and(with_pool(db))
@ -56,3 +60,27 @@ fn with_api_key() -> impl Filter<Extract = (String,), Error = Rejection> + Clone
fn with_pool(db: Pool) -> impl Filter<Extract = (Pool,), Error = Infallible> + Clone { fn with_pool(db: Pool) -> impl Filter<Extract = (Pool,), Error = Infallible> + Clone {
warp::any().map(move || db.clone()) warp::any().map(move || db.clone())
} }
fn with_telem() -> impl Filter<Extract = (crate::Span,), Error = Rejection> + Clone {
warp::any()
.and(warp::header::optional("traceparent"))
.map(|traceparent: Option<String>| {
use opentelemetry::api::trace::{provider::Provider, tracer::Tracer, propagator::HttpTextFormat};
let mut headers = std::collections::HashMap::new();
headers.insert("Traceparent", traceparent.unwrap_or_else(String::new));
let propagator = opentelemetry::api::distributed_context::http_trace_context_propagator::HTTPTraceContextPropagator::new();
let context = propagator.extract(&headers);
if context.is_valid() {
let tracer = opentelemetry::global::trace_provider().get_tracer("api");
let span = tracer.start("context", Some(context));
tracer.mark_span_as_active(&span);
Some(span)
} else {
None
}
})
}

View File

@ -76,8 +76,9 @@ async fn hash_input(form: warp::multipart::FormData) -> (i64, img_hash::ImageHas
(i64::from_be_bytes(buf), hash) (i64::from_be_bytes(buf), hash)
} }
#[tracing::instrument(skip(form, pool, api_key))] #[tracing::instrument(skip(_telem, form, pool, api_key))]
pub async fn search_image( pub async fn search_image(
_telem: crate::Span,
form: warp::multipart::FormData, form: warp::multipart::FormData,
opts: ImageSearchOpts, opts: ImageSearchOpts,
pool: Pool, pool: Pool,
@ -123,8 +124,9 @@ pub async fn search_image(
Ok(warp::reply::json(&similarity)) Ok(warp::reply::json(&similarity))
} }
#[tracing::instrument(skip(form, pool, api_key))] #[tracing::instrument(skip(_telem, form, pool, api_key))]
pub async fn stream_image( pub async fn stream_image(
_telem: crate::Span,
form: warp::multipart::FormData, form: warp::multipart::FormData,
pool: Pool, pool: Pool,
api_key: String, api_key: String,
@ -158,8 +160,9 @@ fn sse_matches(
Ok(warp::sse::json(items)) Ok(warp::sse::json(items))
} }
#[tracing::instrument(skip(form, db, api_key))] #[tracing::instrument(skip(_telem, form, db, api_key))]
pub async fn search_hashes( pub async fn search_hashes(
_telem: crate::Span,
opts: HashSearchOpts, opts: HashSearchOpts,
db: Pool, db: Pool,
api_key: String, api_key: String,
@ -190,8 +193,9 @@ pub async fn search_hashes(
Ok(warp::reply::json(&matches)) Ok(warp::reply::json(&matches))
} }
#[tracing::instrument(skip(db, api_key))] #[tracing::instrument(skip(_telem, db, api_key))]
pub async fn search_file( pub async fn search_file(
_telem: crate::Span,
opts: FileSearchOpts, opts: FileSearchOpts,
db: Pool, db: Pool,
api_key: String, api_key: String,

View File

@ -10,6 +10,8 @@ mod utils;
use warp::Filter; use warp::Filter;
type Span = Option<opentelemetry::global::BoxedSpan>;
fn configure_tracing() { fn configure_tracing() {
use opentelemetry::{ use opentelemetry::{
api::{KeyValue, Provider, Sampler}, api::{KeyValue, Provider, Sampler},
@ -43,7 +45,9 @@ fn configure_tracing() {
}) })
.build(); .build();
let tracer = provider.get_tracer("api"); opentelemetry::global::set_provider(provider);
let tracer = opentelemetry::global::trace_provider().get_tracer("api");
let telem_layer = tracing_opentelemetry::OpentelemetryLayer::with_tracer(tracer); let telem_layer = tracing_opentelemetry::OpentelemetryLayer::with_tracer(tracer);
let fmt_layer = tracing_subscriber::fmt::Layer::default(); let fmt_layer = tracing_subscriber::fmt::Layer::default();