remove connector node from risedev

risedev.yml

@@ -26,9 +26,6 @@ profile:
       # - use: aws-s3
       #   bucket: test-bucket
 
-      # If you want to create CDC source table, uncomment the following line
-      # - use: connector-node
-
       # if you want to enable etcd backend, uncomment the following lines.
       # - use: etcd
       #   unsafe-no-fsync: true
@@ -110,22 +107,6 @@ profile:
       - use: kafka
         persist-data: true
 
-  full-with-connector:
-    steps:
-      - use: minio
-      - use: etcd
-      - use: meta-node
-      - use: compute-node
-      - use: frontend
-      - use: compactor
-      - use: prometheus
-      - use: grafana
-      - use: zookeeper
-        persist-data: true
-      - use: kafka
-        persist-data: true
-      - use: connector-node
-
   standalone-full-peripherals:
     steps:
       - use: minio
@@ -144,7 +125,6 @@ profile:
         persist-data: true
       - use: kafka
         persist-data: true
-      - use: connector-node
 
   standalone-minio-etcd:
     steps:
@@ -1058,9 +1038,6 @@ template:
     # Frontend used by this Prometheus instance
     provide-frontend: "frontend*"
 
-    # Connector-node used by this Prometheus instance
-    provide-connector-node: "connector*"
-
   frontend:
     # Advertise address of frontend
     address: "127.0.0.1"
@@ -1117,19 +1094,6 @@ template:
     # If `user-managed` is true, this service will be started by user with the above config
     user-managed: false
 
-  connector-node:
-    # Connector node advertise address
-    address: "127.0.0.1"
-
-    # Connector node listen port
-    port: 50051
-
-    # Prometheus exporter listen port
-    exporter-port: 50052
-
-    # Id of this instance
-    id: connector-${port}
-
   grafana:
     # Listen address of Grafana
     listen-address: ${address}

src/risedevtool/src/bin/risedev-compose.rs

@@ -222,7 +222,6 @@ fn main() -> Result<()> {
                 (c.address.clone(), c.compose(&compose_config)?)
             }
             ServiceConfig::Redis(_) => return Err(anyhow!("not supported")),
-            ServiceConfig::ConnectorNode(_) => return Err(anyhow!("not supported")),
         };
         compose.container_name = service.id().to_string();
         if opts.deploy {

src/risedevtool/src/bin/risedev-dev.rs

@@ -25,10 +25,10 @@ use indicatif::ProgressBar;
 use risedev::util::{complete_spin, fail_spin};
 use risedev::{
     generate_risedev_env, preflight_check, AwsS3Config, CompactorService, ComputeNodeService,
-    ConfigExpander, ConfigureTmuxTask, ConnectorNodeService, EnsureStopService, ExecuteContext,
-    FrontendService, GrafanaService, KafkaService, MetaNodeService, MinioService, OpendalConfig,
-    PrometheusService, PubsubService, RedisService, ServiceConfig, Task, TempoService,
-    ZooKeeperService, RISEDEV_SESSION_NAME,
+    ConfigExpander, ConfigureTmuxTask, EnsureStopService, ExecuteContext, FrontendService,
+    GrafanaService, KafkaService, MetaNodeService, MinioService, OpendalConfig, PrometheusService,
+    PubsubService, RedisService, ServiceConfig, Task, TempoService, ZooKeeperService,
+    RISEDEV_SESSION_NAME,
 };
 use tempfile::tempdir;
 use yaml_rust::YamlEmitter;
@@ -114,7 +114,6 @@ fn task_main(
             ServiceConfig::AwsS3(_) => None,
             ServiceConfig::OpenDal(_) => None,
             ServiceConfig::RedPanda(_) => None,
-            ServiceConfig::ConnectorNode(c) => Some((c.port, c.id.clone())),
         };
 
         if let Some(x) = listen_info {
@@ -339,17 +338,6 @@ fn task_main(
                 ctx.pb
                     .set_message(format!("redis {}:{}", c.address, c.port));
             }
-            ServiceConfig::ConnectorNode(c) => {
-                let mut ctx =
-                    ExecuteContext::new(&mut logger, manager.new_progress(), status_dir.clone());
-                let mut service = ConnectorNodeService::new(c.clone())?;
-                service.execute(&mut ctx)?;
-                let mut task =
-                    risedev::ConfigureGrpcNodeTask::new(c.address.clone(), c.port, false)?;
-                task.execute(&mut ctx)?;
-                ctx.pb
-                    .set_message(format!("connector grpc://{}:{}", c.address, c.port));
-            }
         }
 
         let service_id = service.id().to_string();

src/risedevtool/src/config.rs

@@ -171,9 +171,6 @@ impl ConfigExpander {
                     "kafka" => ServiceConfig::Kafka(serde_yaml::from_str(&out_str)?),
                     "pubsub" => ServiceConfig::Pubsub(serde_yaml::from_str(&out_str)?),
                     "redis" => ServiceConfig::Redis(serde_yaml::from_str(&out_str)?),
-                    "connector-node" => {
-                        ServiceConfig::ConnectorNode(serde_yaml::from_str(&out_str)?)
-                    }
                     "zookeeper" => ServiceConfig::ZooKeeper(serde_yaml::from_str(&out_str)?),
                     "redpanda" => ServiceConfig::RedPanda(serde_yaml::from_str(&out_str)?),
                     other => return Err(anyhow!("unsupported use type: {}", other)),

src/risedevtool/src/config_gen/prometheus_gen.rs

@@ -79,14 +79,6 @@ impl PrometheusGen {
             .map(|node| format!("\"{}:{}\"", node.address, 9644))
             .join(",");
 
-        let connector_node_targets = config
-            .provide_connector_node
-            .as_ref()
-            .unwrap()
-            .iter()
-            .map(|node| format!("\"{}:{}\"", node.address, node.exporter_port))
-            .join(",");
-
         let now = Local::now().format("%Y%m%d-%H%M%S");
 
         let remote_write = if config.remote_write {
@@ -151,10 +143,6 @@ scrape_configs:
   - job_name: redpanda
     static_configs:
       - targets: [{redpanda_targets}]
-
-  - job_name: connector-node
-    static_configs:
-      - targets: [{connector_node_targets}]
 "#,
         )
     }

src/risedevtool/src/service_config.rs

@@ -190,7 +190,6 @@ pub struct PrometheusConfig {
     pub provide_etcd: Option<Vec<EtcdConfig>>,
     pub provide_redpanda: Option<Vec<RedPandaConfig>>,
     pub provide_frontend: Option<Vec<FrontendConfig>>,
-    pub provide_connector_node: Option<Vec<ConnectorNodeConfig>>,
 }
 
 #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
@@ -318,18 +317,6 @@ pub struct RedisConfig {
     pub address: String,
 }
 
-#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
-#[serde(rename_all = "kebab-case")]
-#[serde(deny_unknown_fields)]
-pub struct ConnectorNodeConfig {
-    #[serde(rename = "use")]
-    phantom_use: Option<String>,
-    pub id: String,
-    pub port: u16,
-    pub exporter_port: u16,
-    pub address: String,
-}
-
 /// All service configuration
 #[derive(Clone, Debug, PartialEq)]
 pub enum ServiceConfig {
@@ -349,7 +336,6 @@ pub enum ServiceConfig {
     Redis(RedisConfig),
     ZooKeeper(ZooKeeperConfig),
     RedPanda(RedPandaConfig),
-    ConnectorNode(ConnectorNodeConfig),
 }
 
 impl ServiceConfig {
@@ -370,7 +356,6 @@ impl ServiceConfig {
             Self::Pubsub(c) => &c.id,
             Self::Redis(c) => &c.id,
             Self::RedPanda(c) => &c.id,
-            Self::ConnectorNode(c) => &c.id,
             Self::OpenDal(c) => &c.id,
         }
     }

src/risedevtool/src/task.rs

@@ -15,7 +15,6 @@
 mod compactor_service;
 mod compute_node_service;
 mod configure_tmux_service;
-mod connector_service;
 mod ensure_stop_service;
 mod etcd_service;
 mod frontend_service;
@@ -52,7 +51,6 @@ pub use utils::*;
 pub use self::compactor_service::*;
 pub use self::compute_node_service::*;
 pub use self::configure_tmux_service::*;
-pub use self::connector_service::*;
 pub use self::ensure_stop_service::*;
 pub use self::etcd_service::*;
 pub use self::frontend_service::*;