chore: add log http ingester scaffold

src/frontend/src/server.rs

@@ -90,6 +90,9 @@ where
  Some(self.instance.clone()),
  );
 
+ builder = builder
+ .with_log_ingest_handler(ServerGrpcQueryHandlerAdapter::arc(self.instance.clone()));
+
  if let Some(user_provider) = self.plugins.get::<UserProviderRef>() {
  builder = builder.with_user_provider(user_provider);
  }

src/servers/Cargo.toml

@@ -104,6 +104,7 @@ tower = { workspace = true, features = ["full"] }
 tower-http = { version = "0.4", features = ["full"] }
 urlencoding = "2.1"
 zstd.workspace = true
+#pipeline = { git = "https://github.com/GreptimeTeam/pipeline.git", rev = "3cdbe92239303b3208ddd71a7ff7630730c91ad5" }
 
 [target.'cfg(not(windows))'.dependencies]
 tikv-jemalloc-ctl = { version = "0.5", features = ["use_std"] }

src/servers/src/http.rs

@@ -65,6 +65,7 @@ use crate::http::prometheus::{
 use crate::metrics::http_metrics_layer;
 use crate::metrics_handler::MetricsHandler;
 use crate::prometheus_handler::PrometheusHandlerRef;
+use crate::query_handler::grpc::ServerGrpcQueryHandlerRef;
 use crate::query_handler::sql::ServerSqlQueryHandlerRef;
 use crate::query_handler::{
  InfluxdbLineProtocolHandlerRef, OpenTelemetryProtocolHandlerRef, OpentsdbProtocolHandlerRef,
@@ -587,6 +588,17 @@ impl HttpServerBuilder {
  }
  }
 
+ // FIXME(qtang): This is a temporary solution to handle the log ingest metrics.
+ pub fn with_log_ingest_handler(self, handler: ServerGrpcQueryHandlerRef) -> Self {
+ Self {
+ router: self.router.nest(
+ &format!("/{HTTP_API_VERSION}/event"),
+ HttpServer::route_log(handler),
+ ),
+ ..self
+ }
+ }
+
  pub fn with_plugins(self, plugins: Plugins) -> Self {
  Self { plugins, ..self }
  }
@@ -699,6 +711,12 @@ impl HttpServer {
  .with_state(metrics_handler)
  }
 
+ fn route_log<S>(grpc_handler: ServerGrpcQueryHandlerRef) -> Router<S> {
+ Router::new()
+ .route("/logs", routing::get(handler::log_ingester))
+ .with_state(grpc_handler)
+ }
+
  fn route_sql<S>(api_state: ApiState) -> ApiRouter<S> {
  ApiRouter::new()
  .api_route(

src/servers/src/http/handler.rs

@@ -17,6 +17,7 @@ use std::env;
 use std::time::Instant;
 
 use aide::transform::TransformOperation;
+use api::v1::RowInsertRequests;
 use axum::extract::{Json, Query, State};
 use axum::response::{IntoResponse, Response};
 use axum::{Extension, Form};
@@ -44,6 +45,7 @@ use crate::http::{
  HttpResponse, ResponseFormat,
 };
 use crate::metrics_handler::MetricsHandler;
+use crate::query_handler::grpc::ServerGrpcQueryHandlerRef;
 use crate::query_handler::sql::ServerSqlQueryHandlerRef;
 
 #[derive(Debug, Default, Serialize, Deserialize, JsonSchema)]
@@ -64,6 +66,30 @@ pub struct SqlQuery {
  pub epoch: Option<String>,
  pub limit: Option<usize>,
 }
+/// handler to log ingester
+#[axum_macros::debug_handler]
+pub async fn log_ingester(
+ State(_state): State<ServerGrpcQueryHandlerRef>,
+ Query(_query_params): Query<SqlQuery>,
+ Extension(_query_ctx): Extension<QueryContextRef>,
+ Json(_payload): Json<Value>,
+) -> String {
+ // TODO (paomian): implement log ingester
+ // transform payload to RowInsertRequest
+ // maybe we need a new trait for log ingester like `LogIngester` instead of `ServerGrpcQueryHandlerRef`
+ // let request = pileline::transform(payload);
+ // state.do_query(payload, query_ctx).await.to_string();
+ let _processors = _payload["processors"].as_str().unwrap();
+ let _data = _payload["data"].as_array().unwrap();
+
+ let insert_request =
+ api::v1::greptime_request::Request::RowInserts(RowInsertRequests::default());
+ let insert_result = _state.do_query(insert_request, _query_ctx).await;
+ match insert_result {
+ Ok(_) => String::from("ok"),
+ Err(e) => e.to_string(),
+ }
+}
 
 /// Handler to execute sql
 #[axum_macros::debug_handler]