feat(telemetry): add object store media type in mete telemetry report (…

…#17553)

proto/telemetry.proto

@@ -99,6 +99,9 @@ message MetaReport {
 
   // How the cluster is deployed
   TelemetryClusterType cluster_type = 7;
+
+  // The object store media type obtained from ObjectStore::store_media_type
+  string object_store_media_type = 8;
 }
 
 enum PlanOptimization {

src/meta/node/src/server.rs

@@ -478,6 +478,7 @@ pub async fn start_service_as_election_leader(
     )
     .await
     .unwrap();
+    let object_store_media_type = hummock_manager.object_store_media_type();
 
     let meta_member_srv = MetaMemberServiceImpl::new(match election_client.clone() {
         None => Either::Right(address_info.clone()),
@@ -742,6 +743,7 @@ pub async fn start_service_as_election_leader(
         Arc::new(MetaReportCreator::new(
             metadata_manager.clone(),
             env.meta_store().backend(),
+            object_store_media_type,
         )),
     );
 

src/meta/src/hummock/manager/mod.rs

@@ -558,6 +558,10 @@ impl HummockManager {
     pub fn metadata_manager(&self) -> &MetadataManager {
         &self.metadata_manager
     }
+
+    pub fn object_store_media_type(&self) -> &'static str {
+        self.object_store.media_type()
+    }
 }
 
 async fn write_exclusive_cluster_id(

src/meta/src/telemetry.rs

@@ -71,6 +71,7 @@ pub struct MetaTelemetryReport {
 
     // Get the ENV from key `TELEMETRY_CLUSTER_TYPE`
     cluster_type: PbTelemetryClusterType,
+    object_store_media_type: &'static str,
 }
 
 impl From<MetaTelemetryJobDesc> for risingwave_pb::telemetry::StreamJobDesc {
@@ -117,6 +118,7 @@ impl TelemetryToProtobuf for MetaTelemetryReport {
             stream_job_count: self.streaming_job_count as u32,
             stream_jobs: self.job_desc.into_iter().map(|job| job.into()).collect(),
             cluster_type: self.cluster_type as i32,
+            object_store_media_type: self.object_store_media_type.to_string(),
         };
         pb_report.encode_to_vec()
     }
@@ -143,13 +145,19 @@ impl TelemetryInfoFetcher for MetaTelemetryInfoFetcher {
 pub struct MetaReportCreator {
     metadata_manager: MetadataManager,
     meta_backend: MetaBackend,
+    object_store_media_type: &'static str,
 }
 
 impl MetaReportCreator {
-    pub fn new(metadata_manager: MetadataManager, meta_backend: MetaBackend) -> Self {
+    pub fn new(
+        metadata_manager: MetadataManager,
+        meta_backend: MetaBackend,
+        object_store_media_type: &'static str,
+    ) -> Self {
         Self {
             metadata_manager,
             meta_backend,
+            object_store_media_type,
         }
     }
 }
@@ -204,6 +212,7 @@ impl TelemetryReportCreator for MetaReportCreator {
             meta_backend: self.meta_backend,
             job_desc: stream_job_desc,
             cluster_type: telemetry_cluster_type_from_env_var(),
+            object_store_media_type: self.object_store_media_type,
         })
     }
 
@@ -255,6 +264,7 @@ mod test {
             },
             job_desc: vec![],
             cluster_type: PbTelemetryClusterType::Unspecified,
+            object_store_media_type: "s3",
         };
 
         let pb_bytes = report.to_pb_bytes();

src/object_store/src/object/mod.rs

@@ -334,6 +334,10 @@ impl ObjectStoreImpl {
     pub fn support_streaming_upload(&self) -> bool {
         dispatch_object_store_enum!(self, |store| store.inner.support_streaming_upload())
     }
+
+    pub fn media_type(&self) -> &'static str {
+        object_store_impl_method_body!(self, media_type())
+    }
 }
 
 fn try_update_failure_metric<T>(