feat: resolve query pool ID at startup
This commit adds a --query-pool flag to router2, used to upsert a
catalog record at startup. Auto-created namespaces will reference this
query pool.
This is for testing only and will be removed in a future commit.
@@ -14,7 +14,6 @@ use crate::{
},
},
};
-use iox_catalog::interface::QueryPoolId;
use observability_deps::tracing::*;
use router2::{
dml_handlers::{NamespaceAutocreation, SchemaValidator, ShardedWriteBuffer},
@@ -68,6 +67,14 @@ pub struct Config {
#[clap(flatten)]
pub(crate) write_buffer_config: WriteBufferConfig,
+
+ /// Query pool name to dispatch writes to.
+ #[clap(
+ long = "--query-pool",
+ env = "INFLUXDB_IOX_QUERY_POOL_NAME",
+ default_value = "iox-shared"
+ )]
+ pub(crate) query_pool_name: String,
}
pub async fn command(config: Config) -> Result<()> {
@@ -87,6 +94,15 @@ pub async fn command(config: Config) -> Result<()> {
let handler_stack =
SchemaValidator::new(write_buffer, Arc::clone(&catalog), Arc::clone(&ns_cache));
+ ////////////////////////////////////////////////////////////////////////////
+ //
+ // THIS CODE IS FOR TESTING ONLY.
+ //
+ // The source of truth for the kafka topics & query pools will be read from
+ // the DB, rather than CLI args for a prod deployment.
+ //
+ ////////////////////////////////////////////////////////////////////////////
+ //
// Look up the kafka topic ID needed to populate namespace creation
// requests.
//
@@ -104,15 +120,27 @@ pub async fn command(config: Config) -> Result<()> {
&config.write_buffer_config.topic
)
});
-
+ let query_id = catalog
+ .query_pools()
+ .create_or_get(&config.query_pool_name)
+ .await
+ .map(|v| v.id)
+ .unwrap_or_else(|e| {
+ panic!(
+ "failed to upsert query pool {} in catalog: {}",
+ &config.write_buffer_config.topic, e
+ )
+ });
let handler_stack = NamespaceAutocreation::new(
catalog,
ns_cache,
topic_id,
- QueryPoolId::new(1),
+ query_id,
"inf".to_owned(),
handler_stack,
);
+ //
+ ////////////////////////////////////////////////////////////////////////////
let http = HttpDelegate::new(config.run_config.max_http_request_size, handler_stack);
let router_server = RouterServer::new(
