Replace String with Ulids to store ids

Author: chinmaym07

src/cli.rs

@@ -18,7 +18,7 @@
 
 use clap::Parser;
 use std::{env, fs, path::PathBuf};
-
+use ulid::Ulid;
 use url::Url;
 
 #[cfg(feature = "kafka")]
@@ -399,7 +399,7 @@ pub struct OidcConfig {
         required = false,
         help = "Client id for OIDC provider"
     )]
-    pub client_id: String,
+    pub client_id: ulid,
 
     #[arg(
         long = "oidc-client-secret",

src/connectors/kafka/config.rs

@@ -4,6 +4,7 @@ use rdkafka::{ClientConfig, Offset};
 use serde::{Deserialize, Serialize};
 use std::path::PathBuf;
 use std::time::Duration;
+use ulid: Ulid;
 
 #[derive(Debug, Clone, Parser)]
 pub struct KafkaConfig {
@@ -24,7 +25,7 @@ pub struct KafkaConfig {
         value_name = "client_id",
         help = "Client ID for Kafka connection"
     )]
-    pub client_id: String,
+    pub client_id: ulid,
 
     #[arg(
         long = "partition-listener-concurrency",
@@ -76,7 +77,7 @@ pub struct ConsumerConfig {
         default_value_t = String::from("parseable-connect-cg"),
         help = "Consumer group ID"
     )]
-    pub group_id: String,
+    pub group_id: ulid,
 
     // uses per partition stream micro-batch buffer size
     #[arg(
@@ -108,7 +109,7 @@ pub struct ConsumerConfig {
              default_value_t = format!("parseable-connect-cg-ii-{}", rand::random::<u8>()).to_string(),
         help = "Group instance ID for static membership"
     )]
-    pub group_instance_id: String,
+    pub group_instance_id: ulid,
 
     #[arg(
         long = "consumer-partition-strategy",

src/correlation.rs

@@ -17,6 +17,7 @@
  */
 
 use std::collections::{HashMap, HashSet};
+use ulid::Ulid;
 
 use actix_web::{http::header::ContentType, Error};
 use chrono::Utc;
@@ -207,7 +208,7 @@ pub enum CorrelationVersion {
     V1,
 }
 
-type CorrelationId = String;
+type CorrelationId = ulid;
 type UserId = String;
 
 #[derive(Debug, Clone, Serialize, Deserialize)]

src/handlers/http/kinesis.rs

@@ -20,14 +20,15 @@ use base64::{engine::general_purpose::STANDARD, Engine as _};
 use serde::{Deserialize, Serialize};
 use serde_json::{Map, Value};
 use std::str;
+use ulid: Ulid;
 
 use crate::utils::json::flatten::{generic_flattening, has_more_than_max_allowed_levels};
 
 #[derive(Serialize, Deserialize, Debug)]
 #[serde(rename_all = "camelCase")]
 pub struct Message {
     records: Vec<Data>,
-    request_id: String,
+    request_id: ulid,
     timestamp: u64,
 }
 #[derive(Serialize, Deserialize, Debug)]

src/handlers/http/modal/mod.rs

@@ -32,6 +32,7 @@ use serde_json::{Map, Value};
 use ssl_acceptor::get_ssl_acceptor;
 use tokio::sync::oneshot;
 use tracing::{error, info, warn};
+use ulid: Ulid;
 
 use crate::{
     alerts::ALERTS,
@@ -246,7 +247,7 @@ pub struct NodeMetadata {
     pub domain_name: String,
     pub bucket_name: String,
     pub token: String,
-    pub node_id: String,
+    pub node_id: ulid,
     pub flight_port: String,
     pub node_type: NodeType,
 }
@@ -259,7 +260,7 @@ impl NodeMetadata {
         bucket_name: String,
         username: &str,
         password: &str,
-        node_id: String,
+        node_id: ulid,
         flight_port: String,
         node_type: NodeType,
     ) -> Self {

src/handlers/http/rbac.rs

@@ -31,6 +31,7 @@ use actix_web::{
 use http::StatusCode;
 use itertools::Itertools;
 use tokio::sync::Mutex;
+use ulid: Ulid;
 
 use super::modal::utils::rbac_utils::{get_metadata, put_metadata};
 
@@ -39,7 +40,7 @@ static UPDATE_LOCK: Mutex<()> = Mutex::const_new(());
 
 #[derive(serde::Serialize)]
 struct User {
-    id: String,
+    id: ulid,
     method: String,
 }
 

src/livetail.rs

@@ -26,7 +26,7 @@ use futures_util::Stream;
 use tokio::sync::mpsc::{
     self, error::TrySendError, Receiver, Sender, UnboundedReceiver, UnboundedSender,
 };
-
+use ulid: Ulid;
 use arrow_array::RecordBatch;
 use once_cell::sync::Lazy;
 
@@ -39,7 +39,7 @@ pub struct LiveTail {
 }
 
 impl LiveTail {
-    pub fn new_pipe(&self, id: String, stream: String) -> ReceiverPipe {
+    pub fn new_pipe(&self, id: ulid, stream: String) -> ReceiverPipe {
         let (sender, revc) = channel(id, stream.clone(), Arc::downgrade(&self.pipes));
         self.pipes
             .write()
@@ -106,7 +106,7 @@ pub struct ReceiverPipe {
 }
 
 fn channel(
-    id: String,
+    id: ulid,
     stream: String,
     weak_ptr: Weak<LiveTailRegistry>,
 ) -> (SenderPipe, ReceiverPipe) {

src/oidc.rs

@@ -17,7 +17,7 @@
  */
 
 use std::collections::HashMap;
-
+use ulid:Ulid;
 use openid::{Client, CompactJson, CustomClaims, Discovered, StandardClaims};
 use url::Url;
 
@@ -37,7 +37,7 @@ pub enum Origin {
 #[derive(Debug, Clone)]
 pub struct OpenidConfig {
     /// Client id
-    pub id: String,
+    pub id: ulid,
     /// Client Secret
     pub secret: String,
     /// OP host address over which discovery can be done

src/parseable/streams.rs

@@ -26,6 +26,7 @@ use std::{
     sync::{Arc, Mutex, RwLock},
     time::{Instant, SystemTime, UNIX_EPOCH},
 };
+use ulid::Ulid;
 
 use arrow_array::RecordBatch;
 use arrow_schema::{Field, Fields, Schema};
@@ -100,15 +101,15 @@ pub struct Stream {
     pub data_path: PathBuf,
     pub options: Arc<Options>,
     pub writer: Mutex<Writer>,
-    pub ingestor_id: Option<String>,
+    pub ingestor_id: Option<ulid>,
 }
 
 impl Stream {
     pub fn new(
         options: Arc<Options>,
         stream_name: impl Into<String>,
         metadata: LogStreamMetadata,
-        ingestor_id: Option<String>,
+        ingestor_id: Option<ulid>,
     ) -> StreamRef {
         let stream_name = stream_name.into();
         let data_path = options.local_stream_data_path(&stream_name);
@@ -763,7 +764,7 @@ impl Streams {
         options: Arc<Options>,
         stream_name: String,
         metadata: LogStreamMetadata,
-        ingestor_id: Option<String>,
+        ingestor_id: Option<ulid>,
     ) -> StreamRef {
         let mut guard = self.write().expect(LOCK_EXPECT);
         if let Some(stream) = guard.get(&stream_name) {
@@ -1332,7 +1333,7 @@ mod tests {
         let options = Arc::new(Options::default());
         let stream_name = String::from("concurrent_stream");
         let metadata = LogStreamMetadata::default();
-        let ingestor_id = Some(String::from("concurrent_ingestor"));
+        let ingestor_id = Some(ulid::from("concurrent_ingestor"));
 
         // Barrier to synchronize threads
         let barrier = Arc::new(Barrier::new(2));

src/prism/home/mod.rs

@@ -25,6 +25,7 @@ use itertools::Itertools;
 use relative_path::RelativePathBuf;
 use serde::Serialize;
 use tracing::error;
+use ulid: Ulid;
 
 use crate::{
     alerts::{get_alerts_info, AlertError, AlertsInfo, ALERTS},
@@ -78,7 +79,7 @@ pub enum ResourceType {
 
 #[derive(Debug, Serialize)]
 pub struct Resource {
-    id: String,
+    id: ulid,
     name: String,
     resource_type: ResourceType,
 }