chore: enhancement of Error Handling

GreptimeTeam · Sep 26, 2024 · 6c9f0d6 · 6c9f0d6
1 parent 2bfc070
commit 6c9f0d6
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 6 deletions.
diff --git a/src/pipeline/src/etl/transform/transformer/greptime.rs b/src/pipeline/src/etl/transform/transformer/greptime.rs
@@ -354,7 +354,7 @@ fn json_value_to_row(schema_info: &mut SchemaInfo, map: Map<String, serde_json::
  Row { values: row }
 }
 
-pub fn identity_pipeline(array: Vec<serde_json::Value>) -> Result<Rows, String> {
+pub fn identity_pipeline(array: Vec<serde_json::Value>) -> Result<Rows> {
  let mut rows = Vec::with_capacity(array.len());
 
  let mut schema = SchemaInfo::default();

diff --git a/src/servers/src/http/event.rs b/src/servers/src/http/event.rs
@@ -437,21 +437,19 @@ async fn ingest_logs_inner(
  for v in pipeline_data {
  pipeline
  .prepare(v, &mut intermediate_state)
- .inspect_err(|reason| {
+ .inspect_err(|_| {
  METRIC_HTTP_LOGS_TRANSFORM_ELAPSED
  .with_label_values(&[db.as_str(), METRIC_FAILURE_VALUE])
  .observe(transform_timer.elapsed().as_secs_f64());
- PipelineTransformSnafu { reason }.build()
  })
  .context(PipelineTransformSnafu)
  .context(PipelineSnafu)?;
  let r = pipeline
  .exec_mut(&mut intermediate_state)
- .inspect_err(|reason| {
+ .inspect_err(|_| {
  METRIC_HTTP_LOGS_TRANSFORM_ELAPSED
  .with_label_values(&[db.as_str(), METRIC_FAILURE_VALUE])
  .observe(transform_timer.elapsed().as_secs_f64());
- PipelineTransformSnafu { reason }.build()
  })
  .context(PipelineTransformSnafu)
  .context(PipelineSnafu)?;
@@ -468,7 +466,7 @@ async fn ingest_logs_inner(
  schema: pipeline.schemas().clone(),
  };
  } else {
- let rows = pipeline::identify_pipeline(pipeline_data)
+ let rows = pipeline::identity_pipeline(pipeline_data)
  .context(PipelineTransformSnafu)
  .context(PipelineSnafu)?;
  transformed_data = rows;