risedev/
risedev_env.rs

1// Copyright 2023 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15#![allow(clippy::doc_markdown)] // RiseDev
16
17use std::fmt::Write;
18use std::process::Command;
19
20use crate::{Application, HummockInMemoryStrategy, ServiceConfig, add_hummock_backend};
21
22/// Generate environment variables (put in file `.risingwave/config/risedev-env`)
23/// from the given service configurations to be used by future
24/// RiseDev commands, like `risedev ctl` or `risedev psql` ().
25pub fn generate_risedev_env(services: &Vec<ServiceConfig>) -> String {
26    let mut env = String::new();
27    for item in services {
28        match item {
29            ServiceConfig::ComputeNode(c) => {
30                // RW_HUMMOCK_URL
31                // If the cluster is launched without a shared storage, we will skip this.
32                {
33                    let mut cmd = Command::new("compute-node");
34                    if add_hummock_backend(
35                        "dummy",
36                        c.provide_opendal.as_ref().unwrap(),
37                        c.provide_minio.as_ref().unwrap(),
38                        c.provide_aws_s3.as_ref().unwrap(),
39                        c.provide_moat.as_ref().unwrap(),
40                        HummockInMemoryStrategy::Disallowed,
41                        &mut cmd,
42                    )
43                    .is_ok()
44                    {
45                        writeln!(
46                            env,
47                            "RW_HUMMOCK_URL=\"{}\"",
48                            cmd.get_args().nth(1).unwrap().to_str().unwrap()
49                        )
50                        .unwrap();
51                    }
52                }
53
54                // RW_META_ADDR
55                {
56                    let meta_node = &c.provide_meta_node.as_ref().unwrap()[0];
57                    writeln!(
58                        env,
59                        "RW_META_ADDR=\"http://{}:{}\"",
60                        meta_node.address, meta_node.port
61                    )
62                    .unwrap();
63                }
64            }
65            ServiceConfig::Frontend(c) => {
66                let listen_address = &c.listen_address;
67                writeln!(
68                    env,
69                    "RISEDEV_RW_FRONTEND_LISTEN_ADDRESS=\"{listen_address}\"",
70                )
71                .unwrap();
72                let port = &c.port;
73                writeln!(env, "RISEDEV_RW_FRONTEND_PORT=\"{port}\"",).unwrap();
74            }
75            ServiceConfig::Kafka(c) => {
76                let brokers = format!("{}:{}", c.address, c.port);
77                writeln!(env, r#"RISEDEV_KAFKA_BOOTSTRAP_SERVERS="{brokers}""#,).unwrap();
78                writeln!(env, r#"RISEDEV_KAFKA_WITH_OPTIONS_COMMON="connector='kafka',properties.bootstrap.server='{brokers}'""#).unwrap();
79                writeln!(env, r#"RPK_BROKERS="{brokers}""#).unwrap();
80            }
81            ServiceConfig::SchemaRegistry(c) => {
82                let url = format!("http://{}:{}", c.address, c.port);
83                writeln!(env, r#"RISEDEV_SCHEMA_REGISTRY_URL="{url}""#,).unwrap();
84                writeln!(env, r#"RPK_REGISTRY_HOSTS="{url}""#).unwrap();
85            }
86            ServiceConfig::Pulsar(c) => {
87                // These 2 names are NOT defined by Pulsar, but by us.
88                // The `pulsar-admin` CLI uses a `PULSAR_CLIENT_CONF` file with `brokerServiceUrl` and `webServiceUrl`
89                // It may be used by our upcoming `PulsarCat` #21401
90                writeln!(
91                    env,
92                    r#"PULSAR_BROKER_URL="pulsar://{}:{}""#,
93                    c.address, c.broker_port
94                )
95                .unwrap();
96                writeln!(
97                    env,
98                    r#"PULSAR_HTTP_URL="http://{}:{}""#,
99                    c.address, c.http_port
100                )
101                .unwrap();
102            }
103            ServiceConfig::Redis(c) => {
104                let host = &c.address;
105                let port = &c.port;
106                let url = format!("redis://{host}:{port}/");
107                writeln!(env, r#"REDIS_HOST="{host}""#).unwrap();
108                writeln!(env, r#"REDIS_PORT="{port}""#).unwrap();
109                writeln!(env, r#"RISEDEV_REDIS_URL="{url}""#).unwrap();
110                if !c.user_managed {
111                    writeln!(env, r#"REDIS_CONTAINER="risedev-{}""#, c.id).unwrap();
112                }
113                writeln!(
114                    env,
115                    r#"RISEDEV_REDIS_WITH_OPTIONS_COMMON="connector='redis',redis.url='{url}'""#,
116                )
117                .unwrap();
118            }
119            ServiceConfig::ClickHouse(c) => {
120                let host = &c.address;
121                let http_port = &c.http_port;
122                let native_port = &c.native_port;
123                let user = &c.user;
124                let password = &c.password;
125                let database = &c.database;
126                let url = format!("http://{host}:{http_port}");
127                writeln!(env, r#"CLICKHOUSE_HOST="{host}""#).unwrap();
128                writeln!(env, r#"CLICKHOUSE_HTTP_PORT="{http_port}""#).unwrap();
129                writeln!(env, r#"CLICKHOUSE_NATIVE_PORT="{native_port}""#).unwrap();
130                writeln!(env, r#"CLICKHOUSE_USER="{user}""#).unwrap();
131                writeln!(env, r#"CLICKHOUSE_PASSWORD="{password}""#).unwrap();
132                writeln!(env, r#"CLICKHOUSE_DATABASE="{database}""#).unwrap();
133                writeln!(env, r#"RISEDEV_CLICKHOUSE_URL="{url}""#).unwrap();
134                if !c.user_managed {
135                    writeln!(env, r#"CLICKHOUSE_CONTAINER="risedev-{}""#, c.id).unwrap();
136                }
137                writeln!(
138                    env,
139                    r#"RISEDEV_CLICKHOUSE_WITH_OPTIONS_COMMON="connector='clickhouse',clickhouse.url='{url}',clickhouse.user='{user}',clickhouse.password='{password}',clickhouse.database='{database}'""#,
140                )
141                .unwrap();
142            }
143            ServiceConfig::MySql(c) if c.application != Application::Metastore => {
144                let host = &c.address;
145                let port = &c.port;
146                let user = &c.user;
147                let password = &c.password;
148                // These envs are used by `mysql` cli.
149                writeln!(env, r#"MYSQL_HOST="{host}""#,).unwrap();
150                writeln!(env, r#"MYSQL_TCP_PORT="{port}""#,).unwrap();
151                // Note: There's no env var for the username read by `mysql` cli. Here we set
152                // `RISEDEV_MYSQL_USER`, which will be read by `e2e_test/commands/mysql` when
153                // running `risedev slt`, as a wrapper of `mysql` cli.
154                writeln!(env, r#"RISEDEV_MYSQL_USER="{user}""#,).unwrap();
155                writeln!(env, r#"MYSQL_PWD="{password}""#,).unwrap();
156                // Note: user and password are not included in the common WITH options.
157                // It's expected to create another dedicated user for the source.
158                writeln!(env, r#"RISEDEV_MYSQL_WITH_OPTIONS_COMMON="connector='mysql-cdc',hostname='{host}',port='{port}'""#,).unwrap();
159            }
160            ServiceConfig::Pubsub(c) => {
161                let address = &c.address;
162                let port = &c.port;
163                writeln!(env, r#"PUBSUB_EMULATOR_HOST="{address}:{port}""#,).unwrap();
164                writeln!(env, r#"RISEDEV_PUBSUB_WITH_OPTIONS_COMMON="connector='google_pubsub',pubsub.emulator_host='{address}:{port}'""#,).unwrap();
165            }
166            ServiceConfig::Postgres(c) => {
167                let host = &c.address;
168                let port = &c.port;
169                let user = &c.user;
170                let password = &c.password;
171                let database = &c.database;
172                // These envs are used by `postgres` cli.
173                writeln!(env, r#"PGHOST="{host}""#,).unwrap();
174                writeln!(env, r#"PGPORT="{port}""#,).unwrap();
175                writeln!(env, r#"PGUSER="{user}""#,).unwrap();
176                writeln!(env, r#"PGPASSWORD="{password}""#,).unwrap();
177                writeln!(env, r#"PGDATABASE="{database}""#,).unwrap();
178                writeln!(
179                    env,
180                    r#"RISEDEV_POSTGRES_WITH_OPTIONS_COMMON="connector='postgres-cdc',hostname='{host}',port='{port}'""#,
181                )
182                .unwrap();
183            }
184            ServiceConfig::Oracle(c) => {
185                let host = &c.address;
186                let port = &c.port;
187                let user = &c.user;
188                let password = &c.password;
189                let database = &c.database;
190                let pdb = &c.pdb;
191                writeln!(env, r#"ORACLE_HOST="{host}""#).unwrap();
192                writeln!(env, r#"ORACLE_PORT="{port}""#).unwrap();
193                writeln!(env, r#"ORACLE_USER="{user}""#).unwrap();
194                writeln!(env, r#"ORACLE_PASSWORD="{password}""#).unwrap();
195                writeln!(env, r#"ORACLE_DATABASE="{database}""#).unwrap();
196                writeln!(env, r#"ORACLE_PDB="{pdb}""#).unwrap();
197                if !c.user_managed {
198                    writeln!(env, r#"ORACLE_CONTAINER="risedev-{}""#, c.id).unwrap();
199                }
200                writeln!(
201                    env,
202                    r#"RISEDEV_ORACLE_WITH_OPTIONS_COMMON="connector='oracle-cdc',hostname='{host}',port='{port}',username='{user}',password='{password}',database.name='{database}',database.pdb.name='{pdb}'""#,
203                )
204                .unwrap();
205            }
206            ServiceConfig::SqlServer(c) => {
207                let host = &c.address;
208                let port = &c.port;
209                let user = &c.user;
210                let password = &c.password;
211                let database = &c.database;
212                // These envs are used by `sqlcmd`.
213                writeln!(env, r#"SQLCMDSERVER="{host}""#,).unwrap();
214                writeln!(env, r#"SQLCMDPORT="{port}""#,).unwrap();
215                writeln!(env, r#"SQLCMDUSER="{user}""#,).unwrap();
216                writeln!(env, r#"SQLCMDPASSWORD="{password}""#,).unwrap();
217                writeln!(env, r#"SQLCMDDBNAME="{database}""#,).unwrap();
218                writeln!(
219                    env,
220                    r#"RISEDEV_SQLSERVER_WITH_OPTIONS_COMMON="connector='sqlserver-cdc',hostname='{host}',port='{port}',username='{user}',password='{password}',database.name='{database}'""#,
221                )
222                .unwrap();
223            }
224            ServiceConfig::MongoDb(c) => {
225                let host = &c.address;
226                let port = &c.port;
227                let url = format!("mongodb://{host}:{port}/?replicaSet=rs0");
228                writeln!(env, r#"MONGODB_HOST="{host}""#).unwrap();
229                writeln!(env, r#"MONGODB_PORT="{port}""#).unwrap();
230                writeln!(env, r#"MONGODB_URL="{url}""#).unwrap();
231                if !c.user_managed {
232                    writeln!(env, r#"MONGODB_CONTAINER="risedev-{}""#, c.id).unwrap();
233                }
234                writeln!(
235                    env,
236                    r#"RISEDEV_MONGODB_WITH_OPTIONS_COMMON="connector='mongodb',mongodb.url='{url}'""#,
237                )
238                .unwrap();
239            }
240            ServiceConfig::ElasticSearch(c) => {
241                let host = &c.address;
242                let port = &c.port;
243                let user = &c.user;
244                let password = &c.password;
245                let url = format!("http://{host}:{port}");
246                writeln!(env, r#"ELASTICSEARCH_HOST="{host}""#).unwrap();
247                writeln!(env, r#"ELASTICSEARCH_PORT="{port}""#).unwrap();
248                writeln!(env, r#"ELASTICSEARCH_USER="{user}""#).unwrap();
249                writeln!(env, r#"ELASTICSEARCH_PASSWORD="{password}""#).unwrap();
250                writeln!(env, r#"RISEDEV_ELASTICSEARCH_URL="{url}""#).unwrap();
251                writeln!(
252                    env,
253                    r#"RISEDEV_ELASTICSEARCH_WITH_OPTIONS_COMMON="connector='elasticsearch',url='{url}',username='{user}',password='{password}'""#,
254                )
255                .unwrap();
256                writeln!(env, r#"SEARCH_SINK_CONNECTOR="elasticsearch""#).unwrap();
257                writeln!(env, r#"SEARCH_SINK_CONNECTION_TYPE="elasticsearch""#).unwrap();
258                writeln!(env, r#"SEARCH_SINK_URL="{url}""#).unwrap();
259                writeln!(env, r#"SEARCH_SINK_USER="{user}""#).unwrap();
260                writeln!(env, r#"SEARCH_SINK_PASSWORD="{password}""#).unwrap();
261                writeln!(
262                    env,
263                    r#"SEARCH_SINK_WITH_OPTIONS_COMMON="connector='elasticsearch',url='{url}',username='{user}',password='{password}'""#,
264                )
265                .unwrap();
266                writeln!(
267                    env,
268                    r#"SEARCH_SINK_ROUTE_WITH_OPTIONS="connector='elasticsearch',connection=es_conn""#,
269                )
270                .unwrap();
271            }
272            ServiceConfig::OpenSearch(c) => {
273                let host = &c.address;
274                let port = &c.port;
275                let user = &c.user;
276                let password = &c.password;
277                let url = format!("http://{host}:{port}");
278                writeln!(env, r#"OPENSEARCH_HOST="{host}""#).unwrap();
279                writeln!(env, r#"OPENSEARCH_PORT="{port}""#).unwrap();
280                writeln!(env, r#"OPENSEARCH_USER="{user}""#).unwrap();
281                writeln!(env, r#"OPENSEARCH_PASSWORD="{password}""#).unwrap();
282                writeln!(env, r#"RISEDEV_OPENSEARCH_URL="{url}""#).unwrap();
283                writeln!(
284                    env,
285                    r#"RISEDEV_OPENSEARCH_WITH_OPTIONS_COMMON="connector='opensearch',url='{url}',username='{user}',password='{password}'""#,
286                )
287                .unwrap();
288                writeln!(env, r#"SEARCH_SINK_CONNECTOR="opensearch""#).unwrap();
289                writeln!(env, r#"SEARCH_SINK_CONNECTION_TYPE="elasticsearch""#).unwrap();
290                writeln!(env, r#"SEARCH_SINK_URL="{url}""#).unwrap();
291                writeln!(env, r#"SEARCH_SINK_USER="{user}""#).unwrap();
292                writeln!(env, r#"SEARCH_SINK_PASSWORD="{password}""#).unwrap();
293                writeln!(
294                    env,
295                    r#"SEARCH_SINK_WITH_OPTIONS_COMMON="connector='opensearch',url='{url}',username='{user}',password='{password}'""#,
296                )
297                .unwrap();
298                writeln!(
299                    env,
300                    r#"SEARCH_SINK_ROUTE_WITH_OPTIONS="connector='opensearch',url='{url}',username='{user}',password='{password}'""#,
301                )
302                .unwrap();
303            }
304            ServiceConfig::Nats(c) => {
305                let server_url = format!("nats://{}:{}", c.address, c.port);
306                writeln!(env, r#"NATS_SERVER_URL="{server_url}""#).unwrap();
307                writeln!(env, r#"RISEDEV_NATS_SERVER_URL="{server_url}""#).unwrap();
308                writeln!(
309                    env,
310                    r#"RISEDEV_NATS_WITH_OPTIONS_COMMON="connector='nats',server_url='{server_url}'""#,
311                )
312                .unwrap();
313            }
314            ServiceConfig::Mqtt(c) => {
315                let url = format!("tcp://{}:{}", c.address, c.port);
316                writeln!(env, r#"MQTT_URL="{url}""#).unwrap();
317                writeln!(env, r#"RISEDEV_MQTT_URL="{url}""#).unwrap();
318                writeln!(
319                    env,
320                    r#"RISEDEV_MQTT_WITH_OPTIONS_COMMON="connector='mqtt',url='{url}'""#,
321                )
322                .unwrap();
323            }
324            ServiceConfig::Minio(c) => {
325                let endpoint = format!("http://{}:{}", c.address, c.port);
326                writeln!(env, r#"RISEDEV_MINIO_ENDPOINT="{endpoint}""#).unwrap();
327                writeln!(env, r#"RISEDEV_MINIO_ACCESS_KEY="{0}""#, c.root_user).unwrap();
328                writeln!(env, r#"RISEDEV_MINIO_SECRET_KEY="{0}""#, c.root_password).unwrap();
329
330                writeln!(
331                    env,
332                    r#"RISEDEV_DELTALAKE_LOCATION="s3a://deltalake/deltalake-test""#
333                )
334                .unwrap();
335                writeln!(env, r#"RISEDEV_DELTALAKE_S3_ENDPOINT="{endpoint}""#).unwrap();
336                writeln!(env, r#"RISEDEV_DELTALAKE_S3_REGION="us-east-1""#).unwrap();
337                writeln!(env, r#"RISEDEV_DELTALAKE_S3_ACCESS_KEY="{0}""#, c.root_user).unwrap();
338                writeln!(
339                    env,
340                    r#"RISEDEV_DELTALAKE_S3_SECRET_KEY="{0}""#,
341                    c.root_password
342                )
343                .unwrap();
344            }
345            ServiceConfig::MetaNode(meta_node_config) => {
346                writeln!(
347                    env,
348                    r#"RISEDEV_RW_META_DASHBOARD_ADDR="http://{}:{}""#,
349                    meta_node_config.address, meta_node_config.dashboard_port
350                )
351                .unwrap();
352            }
353            ServiceConfig::Lakekeeper(c) => {
354                let base_url = format!("http://{}:{}", c.address, c.port);
355                let catalog_url = format!("{}/catalog", base_url);
356                writeln!(env, r#"RISEDEV_LAKEKEEPER_URL="{base_url}""#,).unwrap();
357                writeln!(env, r#"LAKEKEEPER_CATALOG_URL="{catalog_url}""#,).unwrap();
358                writeln!(env, r#"RISEDEV_LAKEKEEPER_WITH_OPTIONS_COMMON="connector='iceberg',catalog.type='rest',catalog.uri='{catalog_url}'""#,).unwrap();
359            }
360            ServiceConfig::Moto(c) => {
361                let endpoint = format!("http://{}:{}", c.address, c.port);
362                let glue_access_key = "my_access_id";
363                let glue_secret_key = "my_secret_key";
364                writeln!(env, r#"RISEDEV_MOTO_ENDPOINT="{endpoint}""#).unwrap();
365                writeln!(env, r#"RISEDEV_GLUE_ENDPOINT="{endpoint}""#).unwrap();
366                writeln!(env, r#"RW_TEST_GLUE_ENDPOINT="{endpoint}""#).unwrap();
367                writeln!(env, r#"RW_TEST_GLUE_ACCESS_KEY="{glue_access_key}""#).unwrap();
368                writeln!(env, r#"RW_TEST_GLUE_SECRET_KEY="{glue_secret_key}""#).unwrap();
369
370                if let Some(minio_configs) = &c.provide_minio
371                    && let Some(minio) = minio_configs.first()
372                {
373                    let s3_endpoint = format!("http://{}:{}", minio.address, minio.port);
374                    let warehouse_path = format!("s3://{}/iceberg", minio.hummock_bucket);
375                    writeln!(env, r#"RW_TEST_GLUE_S3_ENDPOINT="{s3_endpoint}""#).unwrap();
376                    writeln!(env, r#"RW_TEST_GLUE_S3_ACCESS_KEY="{0}""#, minio.root_user).unwrap();
377                    writeln!(
378                        env,
379                        r#"RW_TEST_GLUE_S3_SECRET_KEY="{0}""#,
380                        minio.root_password
381                    )
382                    .unwrap();
383                    writeln!(env, r#"RW_TEST_GLUE_WAREHOUSE_PATH="{warehouse_path}""#).unwrap();
384                    writeln!(env, r#"RISEDEV_GLUE_RUST_WITH_OPTIONS_COMMON="connector='iceberg',catalog.type='glue_rust',catalog.uri='{endpoint}',glue.endpoint='{endpoint}',glue.region='us-east-1',glue.access.key='{glue_access_key}',glue.secret.key='{glue_secret_key}',warehouse.path='{warehouse_path}',s3.endpoint='{s3_endpoint}',s3.region='us-east-1',s3.access.key='{0}',s3.secret.key='{1}',s3.path.style.access='true'""#, minio.root_user, minio.root_password).unwrap();
385                    writeln!(env, r#"RISEDEV_GLUE_JNI_WITH_OPTIONS_COMMON="connector='iceberg',catalog.type='glue',catalog.uri='{endpoint}',glue.endpoint='{endpoint}',glue.region='us-east-1',glue.access.key='{glue_access_key}',glue.secret.key='{glue_secret_key}',warehouse.path='{warehouse_path}',s3.endpoint='{s3_endpoint}',s3.region='us-east-1',s3.access.key='{0}',s3.secret.key='{1}',s3.path.style.access='true'""#, minio.root_user, minio.root_password).unwrap();
386                }
387            }
388            _ => {}
389        }
390    }
391    env
392}
risedev/risedev_env.rs

risedev/
risedev_env.rs
