anna/async-graphql
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Add tracing extension

Browse Source
This commit is contained in:
sunli 2020-04-28 15:01:19 +08:00
parent c4b9b33f5f
commit 495df5a9ab
9 changed files with 273 additions and 139 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
Cargo.toml
View File

@ -41,6 +41,7 @@ mime = "0.3.16"
http = "0.2.1"
fnv = "1.0.6"
regex = "1.3.5"
tracing = "0.1.13"
bson = { version = "0.14.1", optional = true }
uuid = { version = "0.8.1", optional = true }
url = { version = "2.1.1", optional = true }

16
async-graphql-derive/src/object.rs
View File

@ -371,15 +371,13 @@ pub fn generate(object_args: &args::Object, item_impl: &mut ItemImpl) -> Result<
{
method.attrs.remove(idx);
}
} else {
if let Some((idx, _)) = method
.attrs
.iter()
.enumerate()
.find(|(_, a)| a.path.is_ident("field"))
{
method.attrs.remove(idx);
}
} else if let Some((idx, _)) = method
.attrs
.iter()
.enumerate()
.find(|(_, a)| a.path.is_ident("field"))
{
method.attrs.remove(idx);
}
}
}

55
src/context.rs
View File

@ -222,12 +222,42 @@ impl<'a> QueryPathNode<'a> {
}
}
/// Represents the unique id of the resolve
#[derive(Copy, Clone, Debug)]
pub struct ResolveId {
/// Parent id
pub parent: Option<usize>,
/// Current id
pub current: usize,
}
impl ResolveId {
pub(crate) fn root() -> ResolveId {
ResolveId {
parent: None,
current: 0,
}
}
}
impl std::fmt::Display for ResolveId {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
if let Some(parent) = self.parent {
write!(f, "{}:{}", parent, self.current)
} else {
write!(f, "{}", self.current)
}
}
}
/// Query context
#[derive(Clone)]
pub struct ContextBase<'a, T> {
#[allow(missing_docs)]
pub path_node: Option<QueryPathNode<'a>>,
pub(crate) resolve_id: &'a AtomicUsize,
pub(crate) resolve_id: ResolveId,
pub(crate) inc_resolve_id: &'a AtomicUsize,
pub(crate) extensions: &'a [BoxExtension],
pub(crate) item: T,
pub(crate) variables: &'a Variables,
@ -261,11 +291,12 @@ impl Environment {
schema: &'a Schema<Query, Mutation, Subscription>,
path_node: Option<QueryPathNode<'a>>,
item: T,
resolve_id: &'a AtomicUsize,
inc_resolve_id: &'a AtomicUsize,
) -> ContextBase<'a, T> {
ContextBase {
path_node,
resolve_id,
resolve_id: ResolveId::root(),
inc_resolve_id,
extensions: &[],
item,
variables: &self.variables,
@ -279,10 +310,15 @@ impl Environment {
}
impl<'a, T> ContextBase<'a, T> {
#[doc(hidden)]
pub fn get_resolve_id(&self) -> usize {
self.resolve_id
fn get_child_resolve_id(&self) -> ResolveId {
let id = self
.inc_resolve_id
.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
+ 1;
ResolveId {
parent: Some(self.resolve_id.current),
current: id,
}
}
#[doc(hidden)]
@ -299,7 +335,8 @@ impl<'a, T> ContextBase<'a, T> {
}),
extensions: self.extensions,
item: field,
resolve_id: self.resolve_id,
resolve_id: self.get_child_resolve_id(),
inc_resolve_id: self.inc_resolve_id,
variables: self.variables,
variable_definitions: self.variable_definitions,
registry: self.registry,
@ -319,6 +356,7 @@ impl<'a, T> ContextBase<'a, T> {
extensions: self.extensions,
item: selection_set,
resolve_id: self.resolve_id,
inc_resolve_id: &self.inc_resolve_id,
variables: self.variables,
variable_definitions: self.variable_definitions,
registry: self.registry,
@ -460,7 +498,8 @@ impl<'a> ContextBase<'a, &'a SelectionSet> {
}),
extensions: self.extensions,
item: self.item,
resolve_id: self.resolve_id,
resolve_id: self.get_child_resolve_id(),
inc_resolve_id: self.inc_resolve_id,
variables: self.variables,
variable_definitions: self.variable_definitions,
registry: self.registry,

131
src/extensions/apollo_tracing.rs Normal file
View File

@ -0,0 +1,131 @@
use crate::context::ResolveId;
use crate::extensions::{Extension, ResolveInfo};
use chrono::{DateTime, Utc};
use parking_lot::Mutex;
use serde::ser::SerializeMap;
use serde::{Serialize, Serializer};
use std::collections::BTreeMap;
use std::ops::Deref;
struct PendingResolve {
path: serde_json::Value,
field_name: String,
parent_type: String,
return_type: String,
start_time: DateTime<Utc>,
}
struct ResolveStat {
pending_resolve: PendingResolve,
end_time: DateTime<Utc>,
start_offset: i64,
}
impl Deref for ResolveStat {
type Target = PendingResolve;
fn deref(&self) -> &Self::Target {
&self.pending_resolve
}
}
impl Serialize for ResolveStat {
fn serialize<S: Serializer>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error> {
let mut map = serializer.serialize_map(None)?;
map.serialize_entry("path", &self.path)?;
map.serialize_entry("fieldName", &self.field_name)?;
map.serialize_entry("parentType", &self.parent_type)?;
map.serialize_entry("returnType", &self.return_type)?;
map.serialize_entry("startOffset", &self.start_offset)?;
map.serialize_entry(
"duration",
&(self.end_time - self.start_time).num_nanoseconds(),
)?;
map.end()
}
}
struct Inner {
start_time: DateTime<Utc>,
end_time: DateTime<Utc>,
pending_resolves: BTreeMap<usize, PendingResolve>,
resolves: Vec<ResolveStat>,
}
impl Default for Inner {
fn default() -> Self {
Self {
start_time: Utc::now(),
end_time: Utc::now(),
pending_resolves: Default::default(),
resolves: Default::default(),
}
}
}
/// Apollo tracing extension for performance tracing
///
/// Apollo Tracing works by including data in the extensions field of the GraphQL response, which is reserved by the GraphQL spec for extra information that a server wants to return. That way, you have access to performance traces alongside the data returned by your query.
/// It’s already supported by `Apollo Engine`, and we’re excited to see what other kinds of integrations people can build on top of this format.
#[derive(Default)]
pub struct ApolloTracing {
inner: Mutex<Inner>,
}
impl Extension for ApolloTracing {
fn name(&self) -> Option<&'static str> {
Some("tracing")
}
fn parse_start(&self, _query_source: &str) {
self.inner.lock().start_time = Utc::now();
}
fn execution_end(&self) {
self.inner.lock().end_time = Utc::now();
}
fn resolve_field_start(&self, info: &ResolveInfo<'_>) {
let mut inner = self.inner.lock();
inner.pending_resolves.insert(
info.resolve_id.current,
PendingResolve {
path: info.path_node.to_json(),
field_name: info.path_node.field_name().to_string(),
parent_type: info.parent_type.to_string(),
return_type: info.return_type.to_string(),
start_time: Utc::now(),
},
);
}
fn resolve_field_end(&self, resolve_id: ResolveId) {
let mut inner = self.inner.lock();
if let Some(pending_resolve) = inner.pending_resolves.remove(&resolve_id.current) {
let start_offset = (pending_resolve.start_time - inner.start_time)
.num_nanoseconds()
.unwrap();
inner.resolves.push(ResolveStat {
pending_resolve,
start_offset,
end_time: Utc::now(),
});
}
}
fn result(&self) -> Option<serde_json::Value> {
let mut inner = self.inner.lock();
inner
.resolves
.sort_by(|a, b| a.start_offset.cmp(&b.start_offset));
Some(serde_json::json!({
"version": 1,
"startTime": inner.start_time.to_rfc3339(),
"endTime": inner.end_time.to_rfc3339(),
"duration": (inner.end_time - inner.start_time).num_nanoseconds(),
"execution": {
"resolvers": inner.resolves
}
}))
}
}

20
src/extensions/mod.rs
View File

@ -1,9 +1,11 @@
//! Extensions for schema
mod apollo_tracing;
mod tracing;
use crate::context::QueryPathNode;
pub use tracing::ApolloTracing;
pub use self::tracing::Tracing;
use crate::context::{QueryPathNode, ResolveId};
pub use apollo_tracing::ApolloTracing;
pub(crate) type BoxExtension = Box<dyn Extension>;
@ -11,7 +13,7 @@ pub(crate) type BoxExtension = Box<dyn Extension>;
pub struct ResolveInfo<'a> {
/// Because resolver is concurrent, `Extension::resolve_field_start` and `Extension::resolve_field_end` are
/// not strictly ordered, so each pair is identified by an id.
pub resolve_id: usize,
pub resolve_id: ResolveId,
/// Current path node, You can go through the entire path.
pub path_node: &'a QueryPathNode<'a>,
@ -26,8 +28,10 @@ pub struct ResolveInfo<'a> {
/// Represents a GraphQL extension
#[allow(unused_variables)]
pub trait Extension: Sync + Send + 'static {
/// Extension name.
fn name(&self) -> &'static str;
/// If this extension needs to output data to query results, you need to specify a name.
fn name(&self) -> Option<&'static str> {
None
}
/// Called at the begin of the parse.
fn parse_start(&self, query_source: &str) {}
@ -51,8 +55,10 @@ pub trait Extension: Sync + Send + 'static {
fn resolve_field_start(&self, info: &ResolveInfo<'_>) {}
/// Called at the end of the resolve field.
fn resolve_field_end(&self, resolve_id: usize) {}
fn resolve_field_end(&self, resolve_id: ResolveId) {}
/// Get the results
fn result(&self) -> Option<serde_json::Value>;
fn result(&self) -> Option<serde_json::Value> {
None
}
}

164
src/extensions/tracing.rs
View File

@ -1,130 +1,84 @@
use crate::context::ResolveId;
use crate::extensions::{Extension, ResolveInfo};
use chrono::{DateTime, Utc};
use crate::QueryPathSegment;
use parking_lot::Mutex;
use serde::ser::SerializeMap;
use serde::{Serialize, Serializer};
use std::collections::BTreeMap;
use std::ops::Deref;
struct PendingResolve {
path: serde_json::Value,
field_name: String,
parent_type: String,
return_type: String,
start_time: DateTime<Utc>,
}
struct ResolveStat {
pending_resolve: PendingResolve,
end_time: DateTime<Utc>,
start_offset: i64,
}
impl Deref for ResolveStat {
type Target = PendingResolve;
fn deref(&self) -> &Self::Target {
&self.pending_resolve
}
}
impl Serialize for ResolveStat {
fn serialize<S: Serializer>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error> {
let mut map = serializer.serialize_map(None)?;
map.serialize_entry("path", &self.path)?;
map.serialize_entry("fieldName", &self.field_name)?;
map.serialize_entry("parentType", &self.parent_type)?;
map.serialize_entry("returnType", &self.return_type)?;
map.serialize_entry("startOffset", &self.start_offset)?;
map.serialize_entry(
"duration",
&(self.end_time - self.start_time).num_nanoseconds(),
)?;
map.end()
}
}
use tracing::{span, Id, Level};
#[derive(Default)]
struct Inner {
start_time: DateTime<Utc>,
end_time: DateTime<Utc>,
pending_resolves: BTreeMap<usize, PendingResolve>,
resolves: Vec<ResolveStat>,
root_id: Option<Id>,
fields: BTreeMap<usize, Id>,
}
impl Default for Inner {
/// Tracing extension
///
/// # References
///
/// https://crates.io/crates/tracing
pub struct Tracing {
inner: Mutex<Inner>,
}
impl Default for Tracing {
fn default() -> Self {
Self {
start_time: Utc::now(),
end_time: Utc::now(),
pending_resolves: Default::default(),
resolves: Default::default(),
inner: Default::default(),
}
}
}
/// Apollo tracing extension for performance tracing
///
/// Apollo Tracing works by including data in the extensions field of the GraphQL response, which is reserved by the GraphQL spec for extra information that a server wants to return. That way, you have access to performance traces alongside the data returned by your query.
/// It’s already supported by `Apollo Engine`, and we’re excited to see what other kinds of integrations people can build on top of this format.
#[derive(Default)]
pub struct ApolloTracing {
inner: Mutex<Inner>,
}
impl Extension for ApolloTracing {
fn name(&self) -> &'static str {
"tracing"
}
fn parse_start(&self, _query_source: &str) {
self.inner.lock().start_time = Utc::now();
impl Extension for Tracing {
fn parse_start(&self, query_source: &str) {
let root_span = span!(target: "async-graphql", parent:None, Level::INFO, "query", source = query_source);
if let Some(id) = root_span.id() {
tracing::dispatcher::get_default(|d| d.enter(&id));
self.inner.lock().root_id.replace(id);
}
}
fn execution_end(&self) {
self.inner.lock().end_time = Utc::now();
if let Some(id) = self.inner.lock().root_id.take() {
tracing::dispatcher::get_default(|d| d.exit(&id));
}
}
fn resolve_field_start(&self, info: &ResolveInfo<'_>) {
let mut inner = self.inner.lock();
inner.pending_resolves.insert(
info.resolve_id,
PendingResolve {
path: info.path_node.to_json(),
field_name: info.path_node.field_name().to_string(),
parent_type: info.parent_type.to_string(),
return_type: info.return_type.to_string(),
start_time: Utc::now(),
},
);
}
fn resolve_field_end(&self, resolve_id: usize) {
let mut inner = self.inner.lock();
if let Some(pending_resolve) = inner.pending_resolves.remove(&resolve_id) {
let start_offset = (pending_resolve.start_time - inner.start_time)
.num_nanoseconds()
.unwrap();
inner.resolves.push(ResolveStat {
pending_resolve,
start_offset,
end_time: Utc::now(),
});
let parent_span = info
.resolve_id
.parent
.and_then(|id| inner.fields.get(&id))
.cloned();
let span = match &info.path_node.segment {
QueryPathSegment::Index(idx) => span!(
target: "async-graphql",
parent: parent_span,
Level::INFO,
"field",
index = *idx,
parent_type = info.parent_type,
return_type = info.return_type
),
QueryPathSegment::Name(name) => span!(
target: "async-graphql",
parent: parent_span,
Level::INFO,
"field",
name = name,
parent_type = info.parent_type,
return_type = info.return_type
),
};
if let Some(id) = span.id() {
tracing::dispatcher::get_default(|d| d.enter(&id));
inner.fields.insert(info.resolve_id.current, id);
}
}
fn result(&self) -> Option<serde_json::Value> {
let mut inner = self.inner.lock();
inner
.resolves
.sort_by(|a, b| a.start_offset.cmp(&b.start_offset));
Some(serde_json::json!({
"version": 1,
"startTime": inner.start_time.to_rfc3339(),
"endTime": inner.end_time.to_rfc3339(),
"duration": (inner.end_time - inner.start_time).num_nanoseconds(),
"execution": {
"resolvers": inner.resolves
}
}))
fn resolve_field_end(&self, resolve_id: ResolveId) {
if let Some(id) = self.inner.lock().fields.remove(&resolve_id.current) {
tracing::dispatcher::get_default(|d| d.exit(&id));
}
}
}

5
src/mutation_resolver.rs
View File

@ -49,11 +49,10 @@ fn do_resolve<'a, T: ObjectType + Send + Sync>(
let ctx_field = ctx.with_field(field);
let field_name = ctx_field.result_name().to_string();
let resolve_id = ctx_field.get_resolve_id();
if !ctx_field.extensions.is_empty() {
let resolve_info = ResolveInfo {
resolve_id,
resolve_id: ctx_field.resolve_id,
path_node: ctx_field.path_node.as_ref().unwrap(),
parent_type: &T::type_name(),
return_type: match ctx_field
@ -90,7 +89,7 @@ fn do_resolve<'a, T: ObjectType + Send + Sync>(
ctx_field
.extensions
.iter()
.for_each(|e| e.resolve_field_end(resolve_id));
.for_each(|e| e.resolve_field_end(ctx_field.resolve_id));
}
}
Selection::FragmentSpread(fragment_spread) => {

15
src/query.rs
View File

@ -1,4 +1,4 @@
use crate::context::Data;
use crate::context::{Data, ResolveId};
use crate::error::ParseRequestError;
use crate::mutation_resolver::do_mutation_resolve;
use crate::registry::CacheControl;
@ -168,7 +168,7 @@ impl QueryBuilder {
}
// execute
let resolve_id = AtomicUsize::default();
let inc_resolve_id = AtomicUsize::default();
let mut fragments = HashMap::new();
let (selection_set, variable_definitions, is_query) =
current_operation(&document, self.operation_name.as_deref()).ok_or_else(|| {
@ -187,7 +187,8 @@ impl QueryBuilder {
let ctx = ContextBase {
path_node: None,
resolve_id: &resolve_id,
resolve_id: ResolveId::root(),
inc_resolve_id: &inc_resolve_id,
extensions: &extensions,
item: selection_set,
variables: &self.variables,
@ -212,7 +213,13 @@ impl QueryBuilder {
Some(
extensions
.iter()
.filter_map(|e| e.result().map(|res| (e.name().to_string(), res)))
.filter_map(|e| {
if let Some(name) = e.name() {
e.result().map(|res| (name.to_string(), res))
} else {
None
}
})
.collect::<serde_json::Map<_, _>>(),
)
} else {

5
src/resolver.rs
View File

@ -58,11 +58,10 @@ pub fn collect_fields<'a, T: ObjectType + Send + Sync>(
async move {
let ctx_field = ctx.with_field(field);
let field_name = ctx_field.result_name().to_string();
let resolve_id = ctx_field.get_resolve_id();
if !ctx_field.extensions.is_empty() {
let resolve_info = ResolveInfo {
resolve_id,
resolve_id: ctx_field.resolve_id,
path_node: ctx_field.path_node.as_ref().unwrap(),
parent_type: &T::type_name(),
return_type: match ctx_field
@ -101,7 +100,7 @@ pub fn collect_fields<'a, T: ObjectType + Send + Sync>(
ctx_field
.extensions
.iter()
.for_each(|e| e.resolve_field_end(resolve_id));
.for_each(|e| e.resolve_field_end(ctx_field.resolve_id));
}
Ok(res)