fix: log source format in stream info (#1268)

rename as below -
OtelLogs -> otel-logs
OtelMetrics -> otel-metrics
OtelTraces -> otel-traces
Pmeta -> pmeta
Json -> json
Kinesis -> kinesis

added migration steps to perform the rename for existing streams

Author: nikhilsinhaparseable

src/event/format/mod.rs

@@ -59,19 +59,25 @@ type EventSchema = Vec<Arc<Field>>;
 #[derive(Default, Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Hash)]
 pub enum LogSource {
     // AWS Kinesis sends logs in the format of a json array
+    #[serde(rename = "kinesis")]
     Kinesis,
     // OpenTelemetry sends logs according to the specification as explained here
     // https://github.com/open-telemetry/opentelemetry-proto/tree/v1.0.0/opentelemetry/proto/logs/v1
+    #[serde(rename = "otel-logs")]
     OtelLogs,
     // OpenTelemetry sends traces according to the specification as explained here
     // https://github.com/open-telemetry/opentelemetry-proto/blob/v1.0.0/opentelemetry/proto/trace/v1/trace.proto
+    #[serde(rename = "otel-metrics")]
     OtelMetrics,
     // OpenTelemetry sends traces according to the specification as explained here
     // https://github.com/open-telemetry/opentelemetry-proto/tree/v1.0.0/opentelemetry/proto/metrics/v1
+    #[serde(rename = "otel-traces")]
     OtelTraces,
     // Internal Stream format
+    #[serde(rename = "pmeta")]
     Pmeta,
     #[default]
+    #[serde(rename = "json")]
     // Json object or array
     Json,
     Custom(String),

src/migration/mod.rs

@@ -246,6 +246,7 @@ async fn migrate_stream_metadata(
             stream_metadata_value = stream_metadata_migration::v1_v4(stream_metadata_value);
             stream_metadata_value = stream_metadata_migration::v4_v5(stream_metadata_value, stream);
             stream_metadata_value = stream_metadata_migration::v5_v6(stream_metadata_value);
+
             storage
                 .put_object(&path, to_bytes(&stream_metadata_value))
                 .await?;
@@ -259,6 +260,7 @@ async fn migrate_stream_metadata(
             stream_metadata_value = stream_metadata_migration::v2_v4(stream_metadata_value);
             stream_metadata_value = stream_metadata_migration::v4_v5(stream_metadata_value, stream);
             stream_metadata_value = stream_metadata_migration::v5_v6(stream_metadata_value);
+
             storage
                 .put_object(&path, to_bytes(&stream_metadata_value))
                 .await?;
@@ -272,13 +274,15 @@ async fn migrate_stream_metadata(
             stream_metadata_value = stream_metadata_migration::v3_v4(stream_metadata_value);
             stream_metadata_value = stream_metadata_migration::v4_v5(stream_metadata_value, stream);
             stream_metadata_value = stream_metadata_migration::v5_v6(stream_metadata_value);
+
             storage
                 .put_object(&path, to_bytes(&stream_metadata_value))
                 .await?;
         }
         Some("v4") => {
             stream_metadata_value = stream_metadata_migration::v4_v5(stream_metadata_value, stream);
             stream_metadata_value = stream_metadata_migration::v5_v6(stream_metadata_value);
+
             storage
                 .put_object(&path, to_bytes(&stream_metadata_value))
                 .await?;
@@ -289,7 +293,13 @@ async fn migrate_stream_metadata(
                 .put_object(&path, to_bytes(&stream_metadata_value))
                 .await?;
         }
-        _ => (),
+        _ => {
+            stream_metadata_value =
+                stream_metadata_migration::rename_log_source_v6(stream_metadata_value);
+            storage
+                .put_object(&path, to_bytes(&stream_metadata_value))
+                .await?;
+        }
     }
 
     Ok(stream_metadata_value)