feat(telemetry): add telemetry data point when checking license

src/common/src/lib.rs

@@ -84,6 +84,7 @@ pub use {
 };
 pub mod lru;
 pub mod opts;
+pub mod paid_feature;
 pub mod range;
 pub mod row;
 pub mod sequence;

src/license/src/feature.rs → src/common/src/paid_feature.rs

@@ -12,9 +12,11 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use risingwave_license::{License, LicenseKeyError, LicenseManager, Tier};
+use risingwave_pb::telemetry::PbTelemetryEventStage;
 use thiserror::Error;
 
-use super::{License, LicenseKeyError, LicenseManager, Tier};
+use crate::telemetry::report::report_event_common;
 
 /// Define all features that are available based on the tier of the license.
 ///
@@ -81,6 +83,14 @@ macro_rules! def_feature {
                     )*
                 }
             }
+
+            fn get_feature_name(&self) -> &'static str {
+                match &self {
+                    $(
+                        Self::$name => stringify!($name),
+                    )*
+                }
+            }
         }
     };
 }
@@ -91,9 +101,9 @@ for_all_features!(def_feature);
 #[derive(Debug, Error)]
 pub enum FeatureNotAvailable {
     #[error(
-    "feature {:?} is only available for tier {:?} and above, while the current tier is {:?}\n\n\
+        "feature {:?} is only available for tier {:?} and above, while the current tier is {:?}\n\n\
         Hint: You may want to set a license key with `ALTER SYSTEM SET license_key = '...';` command.",
-    feature, feature.min_tier(), current_tier,
+        feature, feature.min_tier(), current_tier,
     )]
     InsufficientTier {
         feature: Feature,
@@ -110,7 +120,7 @@ pub enum FeatureNotAvailable {
 impl Feature {
     /// Check whether the feature is available based on the current license.
     pub fn check_available(self) -> Result<(), FeatureNotAvailable> {
-        match LicenseManager::get().license() {
+        let check_res = match LicenseManager::get().license() {
             Ok(license) => {
                 if license.tier >= self.min_tier() {
                     Ok(())
@@ -133,6 +143,29 @@ impl Feature {
                     })
                 }
             }
+        };
+
+        // Report the event to telemetry
+        let feature_name = Self::get_feature_name(&self);
+        if !feature_name.eq_ignore_ascii_case("TestPaid") {
+            let mut attr_builder = jsonbb::Builder::<Vec<u8>>::new();
+            attr_builder.begin_object();
+            attr_builder.add_string("success");
+            attr_builder.add_value(jsonbb::ValueRef::Bool(check_res.is_ok()));
+            attr_builder.end_object();
+            let attr = attr_builder.finish();
+
+            report_event_common(
+                PbTelemetryEventStage::Unspecified,
+                feature_name,
+                0,
+                None,
+                None,
+                Some(attr),
+                "paywall".to_string(),
+            );
         }
+
+        check_res
     }
 }

src/connector/src/parser/mod.rs

@@ -1233,7 +1233,7 @@ impl SpecificParserConfig {
                 config.schema_location = if let Some(schema_arn) =
                     format_encode_options_with_secret.get(AWS_GLUE_SCHEMA_ARN_KEY)
                 {
-                    risingwave_common::license::Feature::GlueSchemaRegistry
+                    risingwave_common::paid_feature::Feature::GlueSchemaRegistry
                         .check_available()
                         .map_err(anyhow::Error::from)?;
                     SchemaLocation::Glue {

src/connector/src/sink/big_query.rs

@@ -358,7 +358,7 @@ impl Sink for BigQuerySink {
     }
 
     async fn validate(&self) -> Result<()> {
-        risingwave_common::license::Feature::BigQuerySink
+        risingwave_common::paid_feature::Feature::BigQuerySink
             .check_available()
             .map_err(|e| anyhow::anyhow!(e))?;
         if !self.is_append_only && self.pk_indices.is_empty() {

src/connector/src/sink/clickhouse.rs

@@ -530,7 +530,7 @@ impl Sink for ClickHouseSink {
         let (clickhouse_column, clickhouse_engine) =
             query_column_engine_from_ck(client, &self.config).await?;
         if clickhouse_engine.is_shared_tree() {
-            risingwave_common::license::Feature::ClickHouseSharedEngine
+            risingwave_common::paid_feature::Feature::ClickHouseSharedEngine
                 .check_available()
                 .map_err(|e| anyhow::anyhow!(e))?;
         }

src/connector/src/sink/dynamodb.rs

@@ -88,7 +88,7 @@ impl Sink for DynamoDbSink {
     const SINK_NAME: &'static str = DYNAMO_DB_SINK;
 
     async fn validate(&self) -> Result<()> {
-        risingwave_common::license::Feature::DynamoDbSink
+        risingwave_common::paid_feature::Feature::DynamoDbSink
             .check_available()
             .map_err(|e| anyhow::anyhow!(e))?;
         let client = (self.config.build_client().await)

src/connector/src/sink/remote.rs

@@ -166,7 +166,7 @@ impl<R: RemoteSinkTrait> Sink for RemoteSink<R> {
 
 async fn validate_remote_sink(param: &SinkParam, sink_name: &str) -> ConnectorResult<()> {
     if sink_name == OpenSearchSink::SINK_NAME {
-        risingwave_common::license::Feature::OpenSearchSink
+        risingwave_common::paid_feature::Feature::OpenSearchSink
             .check_available()
             .map_err(|e| anyhow::anyhow!(e))?;
     }

src/connector/src/sink/snowflake.rs

@@ -118,7 +118,7 @@ impl Sink for SnowflakeSink {
     }
 
     async fn validate(&self) -> Result<()> {
-        risingwave_common::license::Feature::SnowflakeSink
+        risingwave_common::paid_feature::Feature::SnowflakeSink
             .check_available()
             .map_err(|e| anyhow::anyhow!(e))?;
         if !self.is_append_only {

src/connector/src/sink/sqlserver.rs

@@ -144,7 +144,7 @@ impl Sink for SqlServerSink {
     const SINK_NAME: &'static str = SQLSERVER_SINK;
 
     async fn validate(&self) -> Result<()> {
-        risingwave_common::license::Feature::SqlServerSink
+        risingwave_common::paid_feature::Feature::SqlServerSink
             .check_available()
             .map_err(|e| anyhow::anyhow!(e))?;
 

src/expr/impl/src/scalar/test_license.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use risingwave_common::license::Feature;
+use risingwave_common::paid_feature::Feature;
 use risingwave_expr::{function, ExprError, Result};
 
 /// A function that checks if the `TestPaid` feature is available.

src/frontend/src/handler/create_secret.rs

@@ -15,7 +15,7 @@
 use pgwire::pg_response::{PgResponse, StatementType};
 use prost::Message;
 use risingwave_common::bail_not_implemented;
-use risingwave_common::license::Feature;
+use risingwave_common::paid_feature::Feature;
 use risingwave_sqlparser::ast::{CreateSecretStatement, SqlOption, Value};
 
 use crate::error::{ErrorCode, Result};

src/frontend/src/handler/create_source.rs

@@ -27,7 +27,7 @@ use risingwave_common::catalog::{
     debug_assert_column_ids_distinct, ColumnCatalog, ColumnDesc, ColumnId, Schema, TableId,
     INITIAL_SOURCE_VERSION_ID, KAFKA_TIMESTAMP_COLUMN_NAME,
 };
-use risingwave_common::license::Feature;
+use risingwave_common::paid_feature::Feature;
 use risingwave_common::secret::LocalSecretManager;
 use risingwave_common::types::DataType;
 use risingwave_connector::parser::additional_columns::{

src/frontend/src/handler/create_table.rs

@@ -26,7 +26,7 @@ use risingwave_common::catalog::{
     CdcTableDesc, ColumnCatalog, ColumnDesc, TableId, TableVersionId, DEFAULT_SCHEMA_NAME,
     INITIAL_TABLE_VERSION_ID,
 };
-use risingwave_common::license::Feature;
+use risingwave_common::paid_feature::Feature;
 use risingwave_common::util::sort_util::{ColumnOrder, OrderType};
 use risingwave_common::util::value_encoding::DatumToProtoExt;
 use risingwave_connector::source::cdc::external::{

src/frontend/src/handler/drop_secret.rs

@@ -13,7 +13,7 @@
 // limitations under the License.
 
 use pgwire::pg_response::StatementType;
-use risingwave_common::license::Feature;
+use risingwave_common::paid_feature::Feature;
 use risingwave_sqlparser::ast::ObjectName;
 
 use crate::catalog::root_catalog::SchemaPath;

src/frontend/src/optimizer/plan_node/utils.rs

@@ -320,7 +320,7 @@ macro_rules! plan_node_name {
     };
 }
 pub(crate) use plan_node_name;
-use risingwave_common::license::Feature;
+use risingwave_common::paid_feature::Feature;
 use risingwave_common::types::{DataType, Interval};
 use risingwave_expr::aggregate::PbAggKind;
 use risingwave_pb::plan_common::as_of::AsOfType;

src/license/src/lib.rs

@@ -13,10 +13,8 @@
 // limitations under the License.
 
 mod cpu;
-mod feature;
 mod key;
 mod manager;
 
-pub use feature::*;
 pub use key::*;
 pub use manager::*;

src/license/src/manager.rs

@@ -61,7 +61,7 @@ pub enum Issuer {
 // TODO(license): Shall we add a version field?
 #[derive(Debug, Clone, Deserialize)]
 #[serde(rename_all = "snake_case")]
-pub(super) struct License {
+pub struct License {
     /// Subject of the license.
     ///
     /// See <https://tools.ietf.org/html/rfc7519#section-4.1.2>.
@@ -171,7 +171,7 @@ impl LicenseManager {
     /// Get the current license if it is valid.
     ///
     /// Since the license can expire, the returned license should not be cached by the caller.
-    pub(super) fn license(&self) -> Result<License, LicenseKeyError> {
+    pub fn license(&self) -> Result<License, LicenseKeyError> {
         let license = self.inner.read().unwrap().license.clone()?;
 
         // Check the expiration time additionally.