Naive implementation of http/json collector endpoint

Note this implementation comes with some major concurrency flaws

fpx/src/api.rs

@@ -48,26 +48,23 @@ pub fn create_api(
     events: ServerEvents,
     store: Store,
     inspector_service: InspectorService,
-) -> axum::Router {
-    let api_router = api_router(base_url, events.clone(), store.clone(), inspector_service);
-
-    axum::Router::new()
-        .nest("/api/", api_router)
-        .fallback(studio::default_handler)
-}
-
-fn api_router(
-    base_url: url::Url,
-    events: ServerEvents,
-    store: Store,
-    inspector_service: InspectorService,
 ) -> axum::Router {
     let api_state = ApiState {
         base_url,
         events,
         store,
         inspector_service,
     };
+    let api_router = api_router();
+
+    axum::Router::new()
+        .route("/v1/traces", post(handlers::otel::trace_collector_handler))
+        .nest("/api/", api_router)
+        .with_state(api_state)
+        .fallback(studio::default_handler)
+}
+
+fn api_router() -> axum::Router<ApiState> {
     axum::Router::new()
         .route(
             "/requests/:id",
@@ -90,5 +87,4 @@ fn api_router(
             get(handlers::spans::span_list_handler),
         )
         .fallback(StatusCode::NOT_FOUND)
-        .with_state(api_state)
 }

fpx/src/api/handlers.rs

@@ -1,5 +1,6 @@
 mod inspect;
 mod inspectors;
+pub mod otel;
 mod requestor;
 mod requests;
 pub mod spans;

fpx/src/api/handlers/otel.rs

@@ -0,0 +1,37 @@
+use crate::api::grpc::extract_trace_ids;
+use crate::data::models::Span;
+use crate::data::Store;
+use crate::events::ServerEvents;
+use crate::models::SpanAdded;
+use axum::extract::State;
+use axum::response::IntoResponse;
+use axum::Json;
+use opentelemetry_proto::tonic::collector::trace::v1::{
+    ExportTraceServiceRequest, ExportTraceServiceResponse,
+};
+
+#[tracing::instrument(skip_all)]
+pub async fn trace_collector_handler(
+    State(store): State<Store>,
+    State(events): State<ServerEvents>,
+    Json(payload): Json<ExportTraceServiceRequest>,
+) -> impl IntoResponse {
+    let trace_ids = extract_trace_ids(&payload);
+
+    let tx = store.start_transaction().await.expect("TODO");
+
+    let spans = Span::from_collector_request(payload);
+    for span in spans {
+        store.span_create(&tx, span).await.expect("TODO");
+    }
+
+    store.commit_transaction(tx).await.expect("TODO");
+
+    events.broadcast(SpanAdded::new(trace_ids).into());
+
+    let message = ExportTraceServiceResponse {
+        partial_success: None,
+    };
+
+    Json(message)
+}

fpx/src/commands/dev.rs

@@ -90,7 +90,7 @@ pub async fn handle_command(args: Args) -> Result<()> {
         .into_future();
     let task2 = tonic::transport::Server::builder()
         .add_service(TraceServiceServer::new(grpc_service))
-        .serve("127.0.0.1:4317".parse()?);
+        .serve("127.0.0.1:4567".parse()?);
 
     select! {
         _ = task1 => {},

scripts/otel_collector/config.yaml

@@ -10,14 +10,22 @@ receivers:
 exporters:
   otlphttp:
     endpoint: http://localhost:6767
+    encoding: json
+    compression: none
+
+  otlp:
+    endpoint: http://localhost:4567
+    tls:
+      insecure: true
+    compression: none
 
   debug:
 
 service:
   pipelines:
     traces:
       receivers: [otlp]
-      exporters: [debug]
+      exporters: [debug, otlphttp, otlp]
 
   telemetry:
     logs:

scripts/otel_collector/start.sh

@@ -7,5 +7,6 @@ SCRIPT_DIR=$( cd -- "$( dirname -- "${BASH_SOURCE[0]}" )" &> /dev/null && pwd )
 docker run \
     -p 4317:4317 \
     -p 4318:4318 \
+    --net=host \
     -v "$SCRIPT_DIR/config.yaml:/etc/otelcol-contrib/config.yaml" \
     otel/opentelemetry-collector-contrib:0.103.1